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

package platform

import (
	"crypto/sha1"
	"encoding/hex"
	"strconv"
	"strings"
	"time"
)

func buildRawFrameSQL(database string, query RawFrameQuery) SQLQuery {
	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{}
	limit := query.Limit
	if limit <= 0 {
		limit = 100
	}
	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 ")
	}
	countText := `SELECT COUNT(*) FROM ` + table
	if len(where) > 0 {
		countText += ` WHERE ` + strings.Join(where, " AND ")
	}
	text += ` ORDER BY ts DESC, frame_id ASC LIMIT ` + strconv.Itoa(limit) + ` OFFSET ` + strconv.Itoa(offset)
	return SQLQuery{Text: text, Args: args, CountText: countText}
}

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 := locationTable(database, query)
	where := make([]string, 0, 4)
	args := []any{}
	if protocol := strings.TrimSpace(query["protocol"]); protocol != "" {
		where = append(where, "protocol = '"+quoteTDengine(strings.ToUpper(protocol))+"'")
	}
	if vin := strings.TrimSpace(query["vin"]); 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"])
	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 ")
	}
	countText := `SELECT COUNT(*) FROM ` + table
	if len(where) > 0 {
		countText += ` WHERE ` + strings.Join(where, " AND ")
	}
	text += ` ORDER BY ts DESC, vin ASC, protocol ASC LIMIT ` + strconv.Itoa(limit) + ` OFFSET ` + strconv.Itoa(offset)
	return SQLQuery{Text: text, Args: args, CountText: countText}
}

func qualifyTDengine(database, table string) string {
	if strings.TrimSpace(database) == "" {
		return table
	}
	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
}