feat(platform): query tdengine history data

vehicle-data-platform/apps/api/internal/app/server.go

@@ -2,6 +2,7 @@ package app
 
 import (
 	"context"
+	"database/sql"
 	"log"
 	"net/http"
 	"time"
@@ -16,11 +17,20 @@ func NewServer(cfg config.Config) http.Handler {
 	if cfg.MySQLDSN != "" {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		db, err := platform.OpenMySQL(ctx, cfg.MySQLDSN)
+		db, err := platform.OpenSQL(ctx, "mysql", cfg.MySQLDSN)
 		if err != nil {
 			log.Printf("production mysql store disabled: %v", err)
 		} else {
-			store = platform.NewProductionStore(db, cfg.TDengineDatabase)
+			var tdengine *sql.DB
+			if cfg.TDengineDSN != "" {
+				tdengine, err = platform.OpenSQL(ctx, cfg.TDengineDriver, cfg.TDengineDSN)
+				if err != nil {
+					log.Printf("production tdengine store disabled: %v", err)
+				} else {
+					log.Printf("production tdengine store enabled")
+				}
+			}
+			store = platform.NewProductionStore(db, tdengine, cfg.TDengineDatabase)
 			log.Printf("production mysql store enabled")
 		}
 	}

vehicle-data-platform/apps/api/internal/config/config.go

@@ -13,6 +13,7 @@ type Config struct {
 	RedisUsername    string
 	RedisPassword    string
 	RedisDB          int
+	TDengineDriver   string
 	TDengineDSN      string
 	TDengineDatabase string
 	CapacityCheckBin string
@@ -28,6 +29,7 @@ func Load() Config {
 		RedisUsername:    os.Getenv("REDIS_USERNAME"),
 		RedisPassword:    os.Getenv("REDIS_PASSWORD"),
 		RedisDB:          envInt("REDIS_DB", 50),
+		TDengineDriver:   env("TDENGINE_DRIVER", "taosWS"),
 		TDengineDSN:      os.Getenv("TDENGINE_DSN"),
 		TDengineDatabase: env("TDENGINE_DATABASE", "lingniu_vehicle_ts"),
 		CapacityCheckBin: os.Getenv("CAPACITY_CHECK_BIN"),

vehicle-data-platform/apps/api/internal/platform/mock_store.go

@@ -67,6 +67,10 @@ func (m *MockStore) RealtimeLocations(_ context.Context, query url.Values) (Page
 }
 
 func (m *MockStore) HistoryLocations(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) {
+	return m.HistoryLocationsFromTDengine(ctx, query)
+}
+
+func (m *MockStore) HistoryLocationsFromTDengine(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) {
 	realtime, _ := m.RealtimeLocations(ctx, query)
 	rows := make([]HistoryLocationRow, 0, len(realtime.Items))
 	for _, row := range realtime.Items {

vehicle-data-platform/apps/api/internal/platform/production_store.go

@@ -6,24 +6,26 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/url"
+	"strconv"
 	"strings"
 	"time"
 )
 
 type ProductionStore struct {
-	db       *sql.DB
-	database string
+	db         *sql.DB
+	tdengine   *sql.DB
+	tdDatabase string
 }
 
-func NewProductionStore(db *sql.DB, tdengineDatabase string) *ProductionStore {
+func NewProductionStore(db *sql.DB, tdengine *sql.DB, tdengineDatabase string) *ProductionStore {
 	if db == nil {
 		panic("production db must not be nil")
 	}
-	return &ProductionStore{db: db, database: tdengineDatabase}
+	return &ProductionStore{db: db, tdengine: tdengine, tdDatabase: tdengineDatabase}
 }
 
-func OpenMySQL(ctx context.Context, dsn string) (*sql.DB, error) {
-	db, err := sql.Open("mysql", dsn)
+func OpenSQL(ctx context.Context, driver, dsn string) (*sql.DB, error) {
+	db, err := sql.Open(driver, dsn)
 	if err != nil {
 		return nil, err
 	}
@@ -139,10 +141,86 @@ func (s *ProductionStore) HistoryLocations(ctx context.Context, query url.Values
 	return Page[HistoryLocationRow]{Items: items, Total: realtime.Total, Limit: realtime.Limit, Offset: realtime.Offset}, nil
 }
 
+func (s *ProductionStore) HistoryLocationsFromTDengine(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) {
+	if s.tdengine == nil {
+		return s.HistoryLocations(ctx, query)
+	}
+	limit, offset := buildLimitOffset(query)
+	tdQuery := map[string]string{
+		"protocol": query.Get("protocol"),
+		"vin":      query.Get("vin"),
+		"limit":    strconv.Itoa(limit),
+		"offset":   strconv.Itoa(offset),
+	}
+	if value := strings.TrimSpace(query.Get("dateFrom")); value != "" {
+		tdQuery["dateFrom"] = value
+	}
+	if value := strings.TrimSpace(query.Get("dateTo")); value != "" {
+		tdQuery["dateTo"] = value
+	}
+	built := buildHistoryLocationSQL(s.tdDatabase, tdQuery)
+	rows, err := s.tdengine.QueryContext(ctx, built.Text, built.Args...)
+	if err != nil {
+		return Page[HistoryLocationRow]{}, err
+	}
+	defer rows.Close()
+	items := make([]HistoryLocationRow, 0)
+	for rows.Next() {
+		var row HistoryLocationRow
+		var ts, receivedAt string
+		var longitude, latitude, speed, mileage sql.NullFloat64
+		if err := rows.Scan(&ts, &row.VIN, &row.Protocol, &longitude, &latitude, &speed, &mileage, &receivedAt); err != nil {
+			return Page[HistoryLocationRow]{}, err
+		}
+		row.Longitude = nullFloat64(longitude)
+		row.Latitude = nullFloat64(latitude)
+		row.SpeedKmh = nullFloat64(speed)
+		row.TotalMileageKm = nullFloat64(mileage)
+		row.DeviceTime = ts
+		row.ServerTime = firstNonEmpty(receivedAt, ts)
+		items = append(items, row)
+	}
+	if err := rows.Err(); err != nil {
+		return Page[HistoryLocationRow]{}, err
+	}
+	return Page[HistoryLocationRow]{Items: items, Total: len(items), Limit: limit, Offset: offset}, nil
+}
+
 func (s *ProductionStore) RawFrames(ctx context.Context, query RawFrameQuery) (Page[RawFrameRow], error) {
-	_ = ctx
-	_ = buildRawFrameSQL(s.database, query)
-	return Page[RawFrameRow]{Items: []RawFrameRow{}, Total: 0, Limit: query.Limit, Offset: query.Offset}, nil
+	if s.tdengine == nil {
+		return Page[RawFrameRow]{Items: []RawFrameRow{}, Total: 0, Limit: query.Limit, Offset: query.Offset}, nil
+	}
+	built := buildRawFrameSQL(s.tdDatabase, query)
+	rows, err := s.tdengine.QueryContext(ctx, built.Text, built.Args...)
+	if err != nil {
+		return Page[RawFrameRow]{}, err
+	}
+	defer rows.Close()
+	items := make([]RawFrameRow, 0)
+	for rows.Next() {
+		var row RawFrameRow
+		var ts, frameID, eventTime, receivedAt, parsedFields, parseStatus, parseError, sourceEndpoint, protocol, vehicleKey, vin, phone string
+		var rawSizeBytes int
+		if err := rows.Scan(&ts, &frameID, &eventTime, &receivedAt, &rawSizeBytes, &parsedFields, &parseStatus, &parseError, &sourceEndpoint, &protocol, &vehicleKey, &vin, &phone); err != nil {
+			return Page[RawFrameRow]{}, err
+		}
+		row.ID = frameID
+		row.VIN = vin
+		row.Protocol = protocol
+		row.FrameType = vehicleKey
+		row.DeviceTime = firstNonEmpty(eventTime, ts)
+		row.ServerTime = firstNonEmpty(receivedAt, ts)
+		row.RawSizeBytes = rawSizeBytes
+		row.ParsedFields = parsedFieldsFromString(parsedFields)
+		if len(query.Fields) > 0 {
+			row.ParsedFields = filterParsedFieldsMap(row.ParsedFields, query.Fields)
+		}
+		items = append(items, row)
+	}
+	if err := rows.Err(); err != nil {
+		return Page[RawFrameRow]{}, err
+	}
+	return Page[RawFrameRow]{Items: items, Total: len(items), Limit: query.Limit, Offset: query.Offset}, nil
 }
 
 func (s *ProductionStore) DailyMileage(ctx context.Context, query url.Values) (Page[DailyMileageRow], error) {
@@ -206,11 +284,11 @@ func (s *ProductionStore) OpsHealth(ctx context.Context) (OpsHealth, error) {
 			{Name: "MySQL realtime", Status: mysqlStatus, Detail: mysqlDetail},
 			{Name: "vehicle_realtime_snapshot", Status: "ok", Detail: "读取实时快照表"},
 			{Name: "vehicle_realtime_location", Status: "ok", Detail: "读取实时位置表"},
-			{Name: "TDengine raw_frames", Status: "warning", Detail: "中台 RAW 查询接口待接 TDengine 驱动"},
+			{Name: "TDengine raw_frames", Status: tdengineStatus(s.tdengine), Detail: tdengineDetail(s.tdengine)},
 		},
 		KafkaLag:         0,
 		RedisOnlineKeys:  redisKeys,
-		TDengineWritable: false,
+		TDengineWritable: s.tdengine != nil,
 		MySQLWritable:    mysqlStatus == "ok",
 	}, nil
 }
@@ -242,3 +320,50 @@ func parsedFieldsFromString(value string) map[string]any {
 	}
 	return fields
 }
+
+func filterParsedFieldsMap(fields map[string]any, names []string) map[string]any {
+	if len(names) == 0 || len(fields) == 0 {
+		return fields
+	}
+	out := make(map[string]any, len(names))
+	for _, name := range names {
+		name = strings.TrimSpace(name)
+		if name == "" {
+			continue
+		}
+		if value, ok := fields[name]; ok {
+			out[name] = value
+		}
+	}
+	return out
+}
+
+func nullFloat64(value sql.NullFloat64) float64 {
+	if value.Valid {
+		return value.Float64
+	}
+	return 0
+}
+
+func firstNonEmpty(values ...string) string {
+	for _, value := range values {
+		if strings.TrimSpace(value) != "" {
+			return value
+		}
+	}
+	return ""
+}
+
+func tdengineStatus(db *sql.DB) string {
+	if db == nil {
+		return "warning"
+	}
+	return "ok"
+}
+
+func tdengineDetail(db *sql.DB) string {
+	if db == nil {
+		return "TDengine 未配置或连接失败，历史查询降级"
+	}
+	return "读取 TDengine raw_frames / vehicle_locations"
+}

vehicle-data-platform/apps/api/internal/platform/query_builders_test.go

@@ -36,10 +36,10 @@ func TestBuildRawFrameSQL(t *testing.T) {
 		VIN:      "VIN001",
 		Limit:    1,
 	})
-	if !strings.Contains(built.Text, "lingniu_vehicle_ts.raw_frames") || !strings.Contains(built.Text, "parsed_fields") {
+	if !strings.Contains(built.Text, "lingniu_vehicle_ts.raw_gb32960_") || !strings.Contains(built.Text, "parsed_fields") {
 		t.Fatalf("SQL = %s", built.Text)
 	}
-	if len(built.Args) != 4 || built.Args[0] != "GB32960" || built.Args[1] != "VIN001" || built.Args[2] != 1 {
+	if len(built.Args) != 0 || !strings.Contains(built.Text, "protocol = 'GB32960'") || !strings.Contains(built.Text, "vin = 'VIN001'") || !strings.Contains(built.Text, "LIMIT 1 OFFSET 0") {
 		t.Fatalf("args = %#v", built.Args)
 	}
 }

vehicle-data-platform/apps/api/internal/platform/service.go

@@ -10,6 +10,7 @@ type Store interface {
 	Vehicles(context.Context, url.Values) (Page[VehicleRow], error)
 	RealtimeLocations(context.Context, url.Values) (Page[RealtimeLocationRow], error)
 	HistoryLocations(context.Context, url.Values) (Page[HistoryLocationRow], error)
+	HistoryLocationsFromTDengine(context.Context, url.Values) (Page[HistoryLocationRow], error)
 	RawFrames(context.Context, RawFrameQuery) (Page[RawFrameRow], error)
 	DailyMileage(context.Context, url.Values) (Page[DailyMileageRow], error)
 	QualityIssues(context.Context, url.Values) (Page[QualityIssueRow], error)
@@ -48,7 +49,7 @@ func (s *Service) RealtimeLocations(ctx context.Context, query url.Values) (Page
 }
 
 func (s *Service) HistoryLocations(ctx context.Context, query url.Values) (Page[HistoryLocationRow], error) {
-	return s.store.HistoryLocations(ctx, query)
+	return s.store.HistoryLocationsFromTDengine(ctx, query)
 }
 
 func (s *Service) RawFrames(ctx context.Context, query RawFrameQuery) (Page[RawFrameRow], error) {

vehicle-data-platform/apps/api/internal/platform/tdengine_queries.go

@@ -1,59 +1,78 @@
 package platform
 
-import "strings"
+import (
+	"crypto/sha1"
+	"encoding/hex"
+	"strconv"
+	"strings"
+	"time"
+)
 
 func buildRawFrameSQL(database string, query RawFrameQuery) SQLQuery {
-	table := qualifyTDengine(database, "raw_frames")
-	where := []string{"1 = 1"}
+	table := rawFrameTable(database, query)
+	where := rawFrameWhere(query)
+	parsedFieldsSelect := "'' AS parsed_fields"
+	if query.IncludeFields || len(query.Fields) > 0 {
+		parsedFieldsSelect = "parsed_json AS parsed_fields"
+	}
 	args := []any{}
-	if query.Protocol != "" {
-		where = append(where, "protocol = ?")
-		args = append(args, query.Protocol)
-	}
-	if query.VIN != "" {
-		where = append(where, "vin = ?")
-		args = append(args, query.VIN)
-	}
-	if query.DateFrom != "" {
-		where = append(where, "ts >= ?")
-		args = append(args, query.DateFrom)
-	}
-	if query.DateTo != "" {
-		where = append(where, "ts <= ?")
-		args = append(args, query.DateTo)
-	}
 	limit := query.Limit
 	if limit <= 0 {
 		limit = 100
 	}
-	args = append(args, limit, query.Offset)
-	return SQLQuery{
-		Text: `SELECT ts, frame_id, event_time, received_at, raw_size_bytes, parsed_fields, parse_status, ` +
-			`parse_error, source_endpoint, protocol, vehicle_key, vin, phone FROM ` + table +
-			` WHERE ` + strings.Join(where, " AND ") + ` ORDER BY ts DESC LIMIT ? OFFSET ?`,
-		Args: args,
+	offset := query.Offset
+	if offset < 0 {
+		offset = 0
 	}
+	text := `SELECT ts, frame_id, event_time, received_at, raw_size_bytes, ` + parsedFieldsSelect +
+		`, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone FROM ` + table
+	if len(where) > 0 {
+		text += ` WHERE ` + strings.Join(where, " AND ")
+	}
+	text += ` ORDER BY ts DESC LIMIT ` + strconv.Itoa(limit) + ` OFFSET ` + strconv.Itoa(offset)
+	return SQLQuery{Text: text, Args: args}
+}
+
+func rawFrameWhere(query RawFrameQuery) []string {
+	where := make([]string, 0, 4)
+	if query.Protocol != "" {
+		where = append(where, "protocol = '"+quoteTDengine(strings.ToUpper(strings.TrimSpace(query.Protocol)))+"'")
+	}
+	if query.VIN != "" {
+		where = append(where, "vin = '"+quoteTDengine(strings.TrimSpace(query.VIN))+"'")
+	}
+	if query.DateFrom != "" {
+		where = append(where, "ts >= '"+quoteTDengine(normalizeTDengineTime(query.DateFrom))+"'")
+	}
+	if query.DateTo != "" {
+		where = append(where, "ts <= '"+quoteTDengine(normalizeTDengineTime(query.DateTo))+"'")
+	}
+	return where
 }
 
 func buildHistoryLocationSQL(database string, query map[string]string) SQLQuery {
-	table := qualifyTDengine(database, "vehicle_locations")
-	where := []string{"1 = 1"}
+	table := locationTable(database, query)
+	where := make([]string, 0, 4)
 	args := []any{}
 	if protocol := strings.TrimSpace(query["protocol"]); protocol != "" {
-		where = append(where, "protocol = ?")
-		args = append(args, protocol)
+		where = append(where, "protocol = '"+quoteTDengine(strings.ToUpper(protocol))+"'")
 	}
 	if vin := strings.TrimSpace(query["vin"]); vin != "" {
-		where = append(where, "vin = ?")
-		args = append(args, vin)
+		where = append(where, "vin = '"+quoteTDengine(vin)+"'")
+	}
+	if dateFrom := strings.TrimSpace(query["dateFrom"]); dateFrom != "" {
+		where = append(where, "ts >= '"+quoteTDengine(normalizeTDengineTime(dateFrom))+"'")
+	}
+	if dateTo := strings.TrimSpace(query["dateTo"]); dateTo != "" {
+		where = append(where, "ts <= '"+quoteTDengine(normalizeTDengineTime(dateTo))+"'")
 	}
 	limit, offset := parseLimitOffset(query["limit"], query["offset"])
-	args = append(args, limit, offset)
-	return SQLQuery{
-		Text: `SELECT ts, vin, protocol, longitude, latitude, speed_kmh, total_mileage_km, event_time, received_at FROM ` +
-			table + ` WHERE ` + strings.Join(where, " AND ") + ` ORDER BY ts DESC LIMIT ? OFFSET ?`,
-		Args: args,
+	text := `SELECT ts, vin, protocol, longitude, latitude, speed_kmh, total_mileage_km, received_at FROM ` + table
+	if len(where) > 0 {
+		text += ` WHERE ` + strings.Join(where, " AND ")
 	}
+	text += ` ORDER BY ts DESC LIMIT ` + strconv.Itoa(limit) + ` OFFSET ` + strconv.Itoa(offset)
+	return SQLQuery{Text: text, Args: args}
 }
 
 func qualifyTDengine(database, table string) string {
@@ -62,3 +81,50 @@ func qualifyTDengine(database, table string) string {
 	}
 	return database + "." + table
 }
+
+func rawFrameTable(database string, query RawFrameQuery) string {
+	protocol := strings.ToUpper(strings.TrimSpace(query.Protocol))
+	vin := strings.TrimSpace(query.VIN)
+	if protocol == "" || vin == "" {
+		return qualifyTDengine(database, "raw_frames")
+	}
+	if protocol == "JT808" {
+		return qualifyTDengine(database, "raw_frames")
+	}
+	return qualifyTDengine(database, "raw_"+strings.ToLower(protocol)+"_"+hash16(vin))
+}
+
+func locationTable(database string, query map[string]string) string {
+	protocol := strings.ToUpper(strings.TrimSpace(query["protocol"]))
+	vin := strings.TrimSpace(query["vin"])
+	if protocol == "" || vin == "" {
+		return qualifyTDengine(database, "vehicle_locations")
+	}
+	return qualifyTDengine(database, "loc_"+strings.ToLower(protocol)+"_"+hash16(vin))
+}
+
+func hash16(value string) string {
+	sum := sha1.Sum([]byte(value))
+	return hex.EncodeToString(sum[:8])
+}
+
+func quoteTDengine(value string) string {
+	return strings.ReplaceAll(value, "'", "''")
+}
+
+func normalizeTDengineTime(value string) string {
+	value = strings.TrimSpace(value)
+	if value == "" {
+		return ""
+	}
+	shanghai := time.FixedZone("Asia/Shanghai", 8*3600)
+	for _, layout := range []string{"2006-01-02T15:04:05", "2006-01-02 15:04:05", "2006-01-02"} {
+		if parsed, err := time.ParseInLocation(layout, value, shanghai); err == nil {
+			return parsed.UTC().Format("2006-01-02 15:04:05")
+		}
+	}
+	if parsed, err := time.Parse(time.RFC3339, value); err == nil {
+		return parsed.UTC().Format("2006-01-02 15:04:05")
+	}
+	return value
+}