refactor(go): simplify history location identity
This commit is contained in:
lingniu
@@ -50,7 +50,7 @@ NATS 部署在 Kafka ECS 内网 `172.17.111.56:4222`。
|
||||
| --- | --- | --- | --- |
|
||||
| `raw_frames` | `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` | RAW 证据层，保存完整解析 JSON |
|
||||
| `raw_frame_payload_chunks` | 分片 payload 列 | 协议和车辆标识 tags | 保存超长 raw/parsed payload |
|
||||
| `vehicle_locations` | `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` | 高频历史位置和总里程查询 |
|
||||
| `vehicle_locations` | `ts`、`event_id`、`frame_id`、`received_at`、`longitude`、`latitude`、`altitude_m`、`speed_kmh`、`direction_deg`、`alarm_flag`、`status_flag`、`total_mileage_km` | `protocol`、`vin` | 高频历史位置和总里程查询；只写 VIN 非空车辆 |
|
||||
|
||||
不应重新出现：
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@
|
||||
| --- | --- | --- | --- |
|
||||
| TDengine | `raw_frames` | raw hex/text、完整 `parsed_json`、解析状态、协议标签、车辆标识标签 | `fields_json` 这类可从 `parsed_json`/业务表再得到的重复字段 |
|
||||
| TDengine | `raw_frame_payload_chunks` | 超长 raw/parsed payload 分片 | 业务查询字段 |
|
||||
| TDengine | `vehicle_locations` | 时间、VIN/协议标签、经纬度、速度、方向、SOC、总里程等位置核心字段 | 完整协议 JSON、注册鉴权信息 |
|
||||
| TDengine | `vehicle_locations` | 时间、VIN/协议标签、经纬度、速度、方向、SOC、总里程等位置核心字段 | 完整协议 JSON、注册鉴权信息、phone/device_id/vehicle_key 兜底标识 |
|
||||
| MySQL | `vehicle_realtime_snapshot` | 每个协议+VIN 的最新事件时间、接收时间、车牌、事件 ID | phone、device、source endpoint、完整 JSON |
|
||||
| MySQL | `vehicle_realtime_location` | 每个协议+VIN 的最新位置核心字段 | raw、parsed JSON、消息头内部字段 |
|
||||
| MySQL | `vehicle_daily_mileage` | 日期、VIN、协议、日里程、首末总里程、样本数 | 临时 vehicle key、泛化 metric key/value、每帧细节、位置点列表 |
|
||||
@@ -45,7 +45,12 @@
|
||||
- 无 VIN 的 JT808 等数据保留在 raw/Redis/session 中用于排查和待绑定，不进入正式车辆指标。
|
||||
- 生产：RDS 已在上线前删除泛化 `vehicle_daily_metric`，改为专用 `vehicle_daily_mileage`。
|
||||
|
||||
4. identity 层只保留两张表。
|
||||
4. `vehicle_locations` 不再接收临时身份兜底字段。
|
||||
- 原因：位置历史是正式车辆时间序列，只按 `protocol + vin` 分表和查询。
|
||||
- 无 VIN 的位置帧仍保留在 `raw_frames.parsed_json`，待 identity/binding 修复后可从 raw 重放补写。
|
||||
- 生产：ECS TDengine `vehicle_locations` 已在上线前重建为 `protocol`、`vin` 两个 tag。
|
||||
|
||||
5. identity 层只保留两张表。
|
||||
- `vehicle_identity_binding`：VIN 与 plate/phone/device_id 的映射，供 808 等协议反查 VIN。
|
||||
- `jt808_registration`：808 注册、鉴权、最新活跃和 VIN 匹配状态。
|
||||
- 生产：RDS 已在上线前删除旧 `vehicle_identity_binding_registration`、`vehicle_identity_bindings`，gateway 启动会自动创建最小 schema。
|
||||
|
||||
@@ -54,15 +54,12 @@ type RawFrameRow struct {
|
||||
}
|
||||
|
||||
type LocationQuery struct {
|
||||
Protocol string
|
||||
VehicleKey string
|
||||
VIN string
|
||||
Phone string
|
||||
DeviceID string
|
||||
DateFrom string
|
||||
DateTo string
|
||||
Limit int
|
||||
Offset int
|
||||
Protocol string
|
||||
VIN string
|
||||
DateFrom string
|
||||
DateTo string
|
||||
Limit int
|
||||
Offset int
|
||||
}
|
||||
|
||||
type LocationRow struct {
|
||||
@@ -79,22 +76,16 @@ type LocationRow struct {
|
||||
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 MileagePointQuery struct {
|
||||
Protocol string
|
||||
VehicleKey string
|
||||
VIN string
|
||||
Phone string
|
||||
DeviceID string
|
||||
DateFrom string
|
||||
DateTo string
|
||||
Limit int
|
||||
Offset int
|
||||
Protocol string
|
||||
VIN string
|
||||
DateFrom string
|
||||
DateTo string
|
||||
Limit int
|
||||
Offset int
|
||||
}
|
||||
|
||||
type MileagePointRow struct {
|
||||
@@ -107,10 +98,7 @@ type MileagePointRow struct {
|
||||
Longitude *float64 `json:"longitude,omitempty"`
|
||||
Latitude *float64 `json:"latitude,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 {
|
||||
@@ -253,10 +241,7 @@ func (r *LocationRepository) Query(ctx context.Context, query LocationQuery) ([]
|
||||
&status,
|
||||
&mileage,
|
||||
&row.Protocol,
|
||||
&row.VehicleKey,
|
||||
&row.VIN,
|
||||
&row.Phone,
|
||||
&row.DeviceID,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -306,10 +291,7 @@ func (r *MileagePointRepository) Query(ctx context.Context, query MileagePointQu
|
||||
&longitude,
|
||||
&latitude,
|
||||
&row.Protocol,
|
||||
&row.VehicleKey,
|
||||
&row.VIN,
|
||||
&row.Phone,
|
||||
&row.DeviceID,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -390,10 +372,7 @@ func normalizeRawFrameQuery(query RawFrameQuery) RawFrameQuery {
|
||||
|
||||
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 = normalizeDateTimeLiteral(query.DateFrom)
|
||||
query.DateTo = normalizeDateTimeLiteral(query.DateTo)
|
||||
if query.Limit <= 0 {
|
||||
@@ -404,10 +383,7 @@ func normalizeLocationQuery(query LocationQuery) LocationQuery {
|
||||
|
||||
func normalizeMileagePointQuery(query MileagePointQuery) MileagePointQuery {
|
||||
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 = normalizeDateTimeLiteral(query.DateFrom)
|
||||
query.DateTo = normalizeDateTimeLiteral(query.DateTo)
|
||||
if query.Limit <= 0 {
|
||||
@@ -556,7 +532,7 @@ func quotedList(values []string) string {
|
||||
|
||||
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
|
||||
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, vin FROM ` + table
|
||||
if len(where) > 0 {
|
||||
sqlText += " WHERE " + strings.Join(where, " AND ")
|
||||
}
|
||||
@@ -574,7 +550,7 @@ func buildLocationCountSQL(table string, query LocationQuery) (string, []any) {
|
||||
|
||||
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
|
||||
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 ")
|
||||
}
|
||||
@@ -649,18 +625,9 @@ func locationWhere(query LocationQuery) []string {
|
||||
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(normalizeDateTimeLiteral(query.DateFrom)) + "'")
|
||||
}
|
||||
@@ -678,18 +645,9 @@ func mileagePointWhere(query MileagePointQuery) []string {
|
||||
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(normalizeDateTimeLiteral(query.DateFrom)) + "'")
|
||||
}
|
||||
@@ -883,15 +841,12 @@ func parseMileagePointQuery(r *http.Request) (MileagePointQuery, error) {
|
||||
return MileagePointQuery{}, err
|
||||
}
|
||||
query := MileagePointQuery{
|
||||
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,
|
||||
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")
|
||||
@@ -910,15 +865,12 @@ func parseLocationQuery(r *http.Request) (LocationQuery, error) {
|
||||
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,
|
||||
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 LocationQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD or YYYY-MM-DD HH:mm:ss")
|
||||
|
||||
@@ -252,7 +252,7 @@ func TestRawFrameHandlerReturnsEmptyItemsArrayWhenNoRows(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLocationHandlerReturnsLocationsByVehicleKey(t *testing.T) {
|
||||
func TestLocationHandlerReturnsLocationsByVIN(t *testing.T) {
|
||||
db, mock, err := sqlmock.New()
|
||||
if err != nil {
|
||||
t.Fatalf("sqlmock.New() error = %v", err)
|
||||
@@ -260,18 +260,18 @@ func TestLocationHandlerReturnsLocationsByVehicleKey(t *testing.T) {
|
||||
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'").
|
||||
mock.ExpectQuery("vin = 'LKLG7C4E3NA774736'").
|
||||
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",
|
||||
"direction_deg", "alarm_flag", "status_flag", "total_mileage_km", "protocol", "vin",
|
||||
}).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", "",
|
||||
"JT808", "LKLG7C4E3NA774736",
|
||||
))
|
||||
|
||||
handler := NewLocationHandler(NewLocationRepository(db, "lingniu_vehicle_ts"))
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/history/locations?vehicleKey=JT808:013307811350&protocol=JT808&limit=1", nil)
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/history/locations?vin=LKLG7C4E3NA774736&protocol=JT808&limit=1", nil)
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
handler.ServeHTTP(response, request)
|
||||
@@ -280,17 +280,22 @@ 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":17`} {
|
||||
for _, want := range []string{`"vin":"LKLG7C4E3NA774736"`, `"longitude":121.07764`, `"total_mileage_km":8792.8`, `"total":17`} {
|
||||
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("location response should not expose %s: %s", legacy, body)
|
||||
}
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatalf("sql expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMileagePointHandlerReturnsMileageByVehicleKey(t *testing.T) {
|
||||
func TestMileagePointHandlerReturnsMileageByVIN(t *testing.T) {
|
||||
db, mock, err := sqlmock.New()
|
||||
if err != nil {
|
||||
t.Fatalf("sqlmock.New() error = %v", err)
|
||||
@@ -298,18 +303,18 @@ func TestMileagePointHandlerReturnsMileageByVehicleKey(t *testing.T) {
|
||||
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, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.vehicle_locations").
|
||||
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", "vehicle_key", "vin", "phone", "device_id",
|
||||
"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", "JT808:013307811350", "", "013307811350", "",
|
||||
"JT808", "LKLG7C4E3NA774736",
|
||||
))
|
||||
|
||||
handler := NewMileagePointHandler(NewMileagePointRepository(db, "lingniu_vehicle_ts"))
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/history/mileage-points?vehicleKey=JT808:013307811350&protocol=JT808&limit=1", nil)
|
||||
request := httptest.NewRequest(http.MethodGet, "/api/history/mileage-points?vin=LKLG7C4E3NA774736&protocol=JT808&limit=1", nil)
|
||||
response := httptest.NewRecorder()
|
||||
|
||||
handler.ServeHTTP(response, request)
|
||||
@@ -318,11 +323,16 @@ 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":19`} {
|
||||
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)
|
||||
}
|
||||
@@ -380,12 +390,12 @@ func TestParseMessageIDSupportsDecimalAndHex(t *testing.T) {
|
||||
|
||||
func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) {
|
||||
sqlText, args := buildMileagePointSQL("lingniu_vehicle_ts.vehicle_locations", MileagePointQuery{
|
||||
Protocol: "JT808",
|
||||
VehicleKey: "JT808:013307811350",
|
||||
DateFrom: "2026-07-02 00:00:00",
|
||||
DateTo: "2026-07-02 23:59:59",
|
||||
Limit: 20,
|
||||
Offset: 5,
|
||||
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)
|
||||
@@ -394,7 +404,7 @@ func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) {
|
||||
"FROM lingniu_vehicle_ts.vehicle_locations",
|
||||
"total_mileage_km IS NOT NULL",
|
||||
"protocol = 'JT808'",
|
||||
"vehicle_key = 'JT808:013307811350'",
|
||||
"vin = 'LKLG7C4E3NA774736'",
|
||||
"ts >= '2026-07-01 16:00:00'",
|
||||
"LIMIT 20 OFFSET 5",
|
||||
} {
|
||||
@@ -402,23 +412,26 @@ func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) {
|
||||
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",
|
||||
VehicleKey: "JT808:013307811350",
|
||||
DateFrom: "2026-07-02 00:00:00",
|
||||
DateTo: "2026-07-02 23:59:59",
|
||||
Limit: 20,
|
||||
Offset: 5,
|
||||
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{
|
||||
"protocol = 'JT808'",
|
||||
"vehicle_key = 'JT808:013307811350'",
|
||||
"vin = 'LKLG7C4E3NA774736'",
|
||||
"ts >= '2026-07-01 16:00:00'",
|
||||
"LIMIT 20 OFFSET 5",
|
||||
} {
|
||||
@@ -426,6 +439,9 @@ func TestBuildLocationSQLUsesLiteralsForTDengine(t *testing.T) {
|
||||
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("location sql should use vin-only identity filters: %s", sqlText)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildRawFrameSQLUsesLiteralsForTDengine(t *testing.T) {
|
||||
|
||||
@@ -61,10 +61,7 @@ func SchemaStatements(database string) []string {
|
||||
total_mileage_km DOUBLE
|
||||
) TAGS (
|
||||
protocol NCHAR(32),
|
||||
vehicle_key NCHAR(64),
|
||||
vin NCHAR(32),
|
||||
phone NCHAR(32),
|
||||
device_id NCHAR(64)
|
||||
vin NCHAR(32)
|
||||
)`,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -67,7 +67,7 @@ func (w *Writer) AppendAll(ctx context.Context, env envelope.FrameEnvelope) erro
|
||||
|
||||
func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope) error {
|
||||
table := tableName("raw", env)
|
||||
if err := w.ensureChild(ctx, table, "raw_frames", env); err != nil {
|
||||
if err := w.ensureRawChild(ctx, table, "raw_frames", env); err != nil {
|
||||
return err
|
||||
}
|
||||
rawHex, rawHexChunks := chunkPayload(env, "raw_hex", env.RawHex)
|
||||
@@ -86,7 +86,7 @@ VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedJSON))))
|
||||
return nil
|
||||
}
|
||||
chunkTable := tableName("chunk", env)
|
||||
if err := w.ensureChild(ctx, chunkTable, "raw_frame_payload_chunks", env); err != nil {
|
||||
if err := w.ensureRawChild(ctx, chunkTable, "raw_frame_payload_chunks", env); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, chunk := range chunks {
|
||||
@@ -100,13 +100,16 @@ VALUES (%s)`, chunkTable, joinLiterals(chunkValues(env, chunk)))); err != nil {
|
||||
}
|
||||
|
||||
func (w *Writer) AppendLocation(ctx context.Context, env envelope.FrameEnvelope) error {
|
||||
if strings.TrimSpace(env.VIN) == "" {
|
||||
return nil
|
||||
}
|
||||
longitude, okLon := floatField(env, envelope.FieldLongitude)
|
||||
latitude, okLat := floatField(env, envelope.FieldLatitude)
|
||||
if !okLon || !okLat {
|
||||
return nil
|
||||
}
|
||||
table := tableName("loc", env)
|
||||
if err := w.ensureChild(ctx, table, "vehicle_locations", env); err != nil {
|
||||
table := locationTableName(env)
|
||||
if err := w.ensureLocationChild(ctx, table, env); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
|
||||
@@ -116,7 +119,7 @@ VALUES (%s)`, table, joinLiterals(locationValues(env, longitude, latitude))))
|
||||
return err
|
||||
}
|
||||
|
||||
func (w *Writer) ensureChild(ctx context.Context, table string, stable string, env envelope.FrameEnvelope) error {
|
||||
func (w *Writer) ensureRawChild(ctx context.Context, table string, stable string, env envelope.FrameEnvelope) error {
|
||||
key := stable + "." + table
|
||||
w.cache.mu.Lock()
|
||||
_, ok := w.cache.seen[key]
|
||||
@@ -145,6 +148,27 @@ func (w *Writer) ensureChild(ctx context.Context, table string, stable string, e
|
||||
return nil
|
||||
}
|
||||
|
||||
func (w *Writer) ensureLocationChild(ctx context.Context, table string, env envelope.FrameEnvelope) error {
|
||||
key := "vehicle_locations." + table
|
||||
w.cache.mu.Lock()
|
||||
_, ok := w.cache.seen[key]
|
||||
w.cache.mu.Unlock()
|
||||
if ok {
|
||||
return nil
|
||||
}
|
||||
statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING vehicle_locations TAGS ('%s', '%s')",
|
||||
table,
|
||||
quote(string(env.Protocol)),
|
||||
quote(strings.TrimSpace(env.VIN)))
|
||||
if _, err := w.exec.ExecContext(ctx, statement); err != nil {
|
||||
return err
|
||||
}
|
||||
w.cache.mu.Lock()
|
||||
w.cache.seen[key] = struct{}{}
|
||||
w.cache.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedJSON string) []any {
|
||||
received := millis(env.ReceivedAtMS)
|
||||
eventTime := millis(env.EventTimeMS)
|
||||
@@ -246,6 +270,10 @@ func tableName(prefix string, env envelope.FrameEnvelope) string {
|
||||
return prefix + "_" + strings.ToLower(string(env.Protocol)) + "_" + hash16(env.VehicleKey())
|
||||
}
|
||||
|
||||
func locationTableName(env envelope.FrameEnvelope) string {
|
||||
return "loc_" + strings.ToLower(string(env.Protocol)) + "_" + hash16(strings.TrimSpace(env.VIN))
|
||||
}
|
||||
|
||||
func normalizePhoneTag(phone string) string {
|
||||
trimmed := strings.TrimLeft(strings.TrimSpace(phone), "0")
|
||||
if trimmed == "" {
|
||||
|
||||
@@ -22,6 +22,12 @@ func TestSchemaStatementsCreateCoreStables(t *testing.T) {
|
||||
if strings.Contains(statements, "fields_json") {
|
||||
t.Fatalf("raw schema should not create duplicate fields_json column:\n%s", statements)
|
||||
}
|
||||
locationStable := between(statements, "CREATE STABLE IF NOT EXISTS vehicle_locations", "}")
|
||||
for _, field := range []string{"vehicle_key", "phone", "device_id"} {
|
||||
if strings.Contains(locationStable, field) {
|
||||
t.Fatalf("location stable should only keep protocol/vin identity tags, found %s:\n%s", field, locationStable)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriterAppendsRawAndLocationOnly(t *testing.T) {
|
||||
@@ -61,6 +67,17 @@ func TestWriterAppendsRawAndLocationOnly(t *testing.T) {
|
||||
if strings.Contains(rawInsert, "fields_json") {
|
||||
t.Fatalf("raw insert should not write duplicate fields_json: %s", rawInsert)
|
||||
}
|
||||
locationChild := findSQL(exec.calls, "USING vehicle_locations")
|
||||
for _, want := range []string{"'JT808'", "'LNBVIN00000000001'"} {
|
||||
if !strings.Contains(locationChild, want) {
|
||||
t.Fatalf("location child should tag by protocol and vin only: %s", locationChild)
|
||||
}
|
||||
}
|
||||
for _, field := range []string{"'JT808:013307795425'", "'13307795425'", "device_id"} {
|
||||
if strings.Contains(locationChild, field) {
|
||||
t.Fatalf("location child should not carry raw identity fallback tag %s: %s", field, locationChild)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriterNormalizesPhoneTagAndRefreshesExistingChildTags(t *testing.T) {
|
||||
@@ -130,6 +147,27 @@ func TestWriterSkipsSparseDerivedRows(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriterSkipsLocationWhenVINIsMissing(t *testing.T) {
|
||||
exec := &recordingExec{}
|
||||
writer := NewWriter(exec)
|
||||
env := sampleEnvelope()
|
||||
env.VIN = ""
|
||||
|
||||
if err := writer.AppendAll(context.Background(), env); err != nil {
|
||||
t.Fatalf("AppendAll() error = %v", err)
|
||||
}
|
||||
|
||||
if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 {
|
||||
t.Fatalf("raw insert count = %d", got)
|
||||
}
|
||||
if got := countSQL(exec.calls, "USING vehicle_locations"); got != 0 {
|
||||
t.Fatalf("location child create count = %d", got)
|
||||
}
|
||||
if got := countSQL(exec.calls, "INSERT INTO loc_"); got != 0 {
|
||||
t.Fatalf("location insert count = %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRawSizeBytesUsesRawTextWhenHexIsEmpty(t *testing.T) {
|
||||
env := envelope.FrameEnvelope{RawText: `{"code":"0F80","data":{"speed":12}}`}
|
||||
if got, want := rawSizeBytes(env), len([]byte(env.RawText)); got != want {
|
||||
@@ -195,6 +233,19 @@ func findSQL(calls []execCall, pattern string) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func between(value string, start string, end string) string {
|
||||
startIndex := strings.Index(value, start)
|
||||
if startIndex < 0 {
|
||||
return ""
|
||||
}
|
||||
value = value[startIndex:]
|
||||
endIndex := strings.Index(value, end)
|
||||
if endIndex < 0 {
|
||||
return value
|
||||
}
|
||||
return value[:endIndex]
|
||||
}
|
||||
|
||||
type execCall struct {
|
||||
query string
|
||||
args []any
|
||||
|
||||
Reference in New Issue
Block a user