laic-backend/service/telemetry_store.go

package service

import (
	"fmt"
	"strconv"
	"strings"
	"sync"
	"time"

	"laic-backend/common"
	"laic-backend/logger"
)

// TelemetryPoint 无人机遥测点（对应 device_telemetry 超级表）
type TelemetryPoint struct {
	Ts          int64   `json:"ts"`
	Longitude   float64 `json:"longitude"`
	Latitude    float64 `json:"latitude"`
	Altitude    float64 `json:"altitude"`
	GroundSpeed float64 `json:"groundSpeed"`
	Roll        float64 `json:"roll"`
	Pitch       float64 `json:"pitch"`
	Yaw         float64 `json:"yaw"`
	BatteryPct  int     `json:"batteryPct"`
	BatteryV    float64 `json:"batteryV"`
	Satellites  int     `json:"satellites"`
	GpsQuality  string  `json:"gpsQuality"`
	LinkQuality int     `json:"linkQuality"`
	FlightMode  string  `json:"flightMode"`
	Armed       int8    `json:"armed"`
}

const (
	telemetryBatchSize     = 500
	telemetryFlushInterval = 10 * time.Second
)

type telemetryRecord struct {
	DockID string
	Point  TelemetryPoint
}

// TelemetryStore 遥测缓冲：累计 500 条或 10 秒后批量写入 TDengine
type TelemetryStore struct {
	mu      sync.Mutex
	buffer  []telemetryRecord
	droneSN sync.Map // dockID -> droneSN（来自 state/drone 上报）
}

var DefaultTelemetryStore = &TelemetryStore{}

// SetDroneSN 记录机巢当前无人机序列号（作为 device_telemetry 的 tag）
func (s *TelemetryStore) SetDroneSN(dockID, sn string) {
	if sn != "" {
		s.droneSN.Store(dockID, sn)
	}
}

// Append 追加一个遥测点，达到阈值立即刷盘
func (s *TelemetryStore) Append(dockID string, p TelemetryPoint) {
	s.mu.Lock()
	s.buffer = append(s.buffer, telemetryRecord{DockID: dockID, Point: p})
	if len(s.buffer) >= telemetryBatchSize {
		s.flushLocked()
	}
	s.mu.Unlock()
}

// Start 启动定时刷盘协程
func (s *TelemetryStore) Start() {
	go func() {
		ticker := time.NewTicker(telemetryFlushInterval)
		defer ticker.Stop()
		for range ticker.C {
			s.mu.Lock()
			s.flushLocked()
			s.mu.Unlock()
		}
	}()
}

func (s *TelemetryStore) flushLocked() {
	if len(s.buffer) == 0 {
		return
	}
	records := s.buffer
	s.buffer = nil
	s.write(records)
}

func (s *TelemetryStore) write(records []telemetryRecord) {
	if common.TD == nil {
		return
	}
	groups := make(map[string][]TelemetryPoint, len(records))
	for _, r := range records {
		groups[r.DockID] = append(groups[r.DockID], r.Point)
	}
	for dockID, points := range groups {
		s.insert(dockID, s.droneSNValue(dockID), points)
	}
}

func (s *TelemetryStore) droneSNValue(dockID string) string {
	if v, ok := s.droneSN.Load(dockID); ok {
		return v.(string)
	}
	return ""
}

// DroneSN 返回机巢当前无人机序列号（来自 state/drone 上报）
func (s *TelemetryStore) DroneSN(dockID string) string {
	return s.droneSNValue(dockID)
}

func (s *TelemetryStore) insert(dockID, droneSN string, points []TelemetryPoint) {
	defer func() {
		if r := recover(); r != nil {
			logger.ERROR("TDengine 写入 panic（已恢复，避免进程崩溃）", fmt.Errorf("%v", r))
		}
	}()
	var b strings.Builder
	b.WriteString("INSERT INTO ")
	b.WriteString(sanitizeTable(dockID))
	b.WriteString(" USING device_telemetry TAGS ('")
	b.WriteString(escapeTD(dockID))
	b.WriteString("', '")
	b.WriteString(escapeTD(droneSN))
	b.WriteString("') VALUES ")
	for i, p := range points {
		if i > 0 {
			b.WriteString(", ")
		}
		ts := p.Ts
		if ts <= 0 {
			ts = time.Now().UnixMilli()
		}
		fmt.Fprintf(&b, "(%d, %s, %s, %s, %s, %s, %s, %s, %d, %s, %d, '%s', %d, '%s', %d)",
			ts,
			ftoa(p.Longitude), ftoa(p.Latitude), ftoa(p.Altitude), ftoa(p.GroundSpeed),
			ftoa(p.Roll), ftoa(p.Pitch), ftoa(p.Yaw),
			p.BatteryPct, ftoa(p.BatteryV), p.Satellites,
			escapeTD(p.GpsQuality), p.LinkQuality, escapeTD(p.FlightMode), p.Armed,
		)
	}
	if _, err := common.TD.Exec(b.String()); err != nil {
		logger.ERROR("TDengine 批量写入遥测失败", err)
	}
}

func ftoa(v float64) string {
	return strconv.FormatFloat(v, 'f', -1, 64)
}

func escapeTD(s string) string {
	return strings.ReplaceAll(s, "'", "\\'")
}

// sanitizeTable 将 dockID 转为 TDengine 子表名（仅保留字母数字）
func sanitizeTable(dockID string) string {
	var b strings.Builder
	b.WriteString("d_")
	for _, r := range dockID {
		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') {
			b.WriteRune(r)
		} else {
			b.WriteByte('_')
		}
	}
	return b.String()
}