laic-backend/service/trajectory_store.go

package service

import (
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"sync"
	"time"

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

// trajectoryBuffer 单次执行的活动轨迹缓冲
type trajectoryBuffer struct {
	ExecutionID int64
	DockID      string
	DroneSN     string
	Points      []vo.TrajectoryPoint
}

// TrajectoryStore 轨迹采集：维护每个机巢的活动执行，遥测点入缓冲，终态回写 MySQL + TDengine
type TrajectoryStore struct {
	mu     sync.Mutex
	active map[string]*trajectoryBuffer // dockID -> 活动执行
}

var DefaultTrajectoryStore = &TrajectoryStore{active: make(map[string]*trajectoryBuffer)}

// Begin 建立活动执行关联（任务下发成功后调用）
func (s *TrajectoryStore) Begin(dockID string, executionID int64, droneSN string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.active[dockID] = &trajectoryBuffer{
		ExecutionID: executionID,
		DockID:      dockID,
		DroneSN:     droneSN,
		Points:      make([]vo.TrajectoryPoint, 0, 1024),
	}
}

// Append 将遥测点追加到活动执行的轨迹缓冲
func (s *TrajectoryStore) Append(dockID string, p TelemetryPoint) {
	s.mu.Lock()
	defer s.mu.Unlock()
	buf, ok := s.active[dockID]
	if !ok {
		return
	}
	ts := p.Ts
	if ts <= 0 {
		ts = time.Now().UnixMilli()
	}
	buf.Points = append(buf.Points, vo.TrajectoryPoint{
		Ts:          ts,
		Longitude:   p.Longitude,
		Latitude:    p.Latitude,
		Altitude:    p.Altitude,
		GroundSpeed: p.GroundSpeed,
		Yaw:         p.Yaw,
		BatteryPct:  p.BatteryPct,
	})
}

// Finalize 终态回写：轨迹点写 MySQL trajectory_json + TDengine task_trajectory，并清除活动关联
func (s *TrajectoryStore) Finalize(dockID string, droneSN string) {
	s.mu.Lock()
	buf, ok := s.active[dockID]
	if ok {
		delete(s.active, dockID)
	}
	s.mu.Unlock()
	if !ok || buf == nil {
		return
	}
	if droneSN != "" {
		buf.DroneSN = droneSN
	}

	pointsJSON, err := json.Marshal(buf.Points)
	if err != nil {
		pointsJSON = []byte("[]")
	}
	updates := map[string]any{"trajectory_json": string(pointsJSON)}
	if buf.DroneSN != "" {
		updates["drone_sn"] = buf.DroneSN
	}
	if err := common.DB.Model(&model.TaskExecution{}).
		Where("id = ?", buf.ExecutionID).
		Updates(updates).Error; err != nil {
		logger.ERROR("回写执行轨迹 JSON 失败", err)
	}

	s.writeTDengine(buf)
}

// writeTDengine 轨迹点批量写入 task_trajectory 超级表（子表按 execution_id 隔离）
func (s *TrajectoryStore) writeTDengine(buf *trajectoryBuffer) {
	if common.TD == nil || len(buf.Points) == 0 {
		return
	}
	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(trajectoryTable(buf.ExecutionID))
	b.WriteString(" USING task_trajectory TAGS (")
	fmt.Fprintf(&b, "%d, '%s', '%s') VALUES ",
		buf.ExecutionID, escapeTD(buf.DockID), escapeTD(buf.DroneSN))
	for i, p := range buf.Points {
		if i > 0 {
			b.WriteString(", ")
		}
		fmt.Fprintf(&b, "(%d, %s, %s, %s, %s, %s, %d)",
			p.Ts, ftoa(p.Longitude), ftoa(p.Latitude), ftoa(p.Altitude),
			ftoa(p.GroundSpeed), ftoa(p.Yaw), p.BatteryPct)
	}
	if _, err := common.TD.Exec(b.String()); err != nil {
		logger.ERROR("TDengine 批量写入轨迹失败", err)
	}
}

// trajectoryTable 按执行 ID 生成子表名
func trajectoryTable(execID int64) string {
	return "tr_" + strconv.FormatInt(execID, 10)
}