laic-backend/service/command_service.go

package service

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
	"gorm.io/gorm"

	"laic-backend/cache"
	"laic-backend/common"
	"laic-backend/logger"
	"laic-backend/model"
	"laic-backend/mqtt"
	"laic-backend/tool"
	"laic-backend/vo"
)

type CommandService struct{}

var DefaultCommandService = &CommandService{}

const (
	maxRetryCount           = 2
	workflowStartTimeout    = 2 * time.Minute
	workflowProgressTimeout = 5 * time.Minute
)

// Dispatch 下发指令：校验在线 → 落库 → MQTT 发布
func (s *CommandService) Dispatch(userID int64, isAdmin bool, id int64, req *vo.CommandReq) (*model.DeviceCommandLog, *common.BusiError) {
	var dock model.Dock
	if err := common.DB.Scopes(withUserFilter(userID, isAdmin)).First(&dock, id).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.ErrDockNotFound
		}
		return nil, common.ErrInternal
	}
	dockID := dock.DockID

	online, _ := common.SetMemberExists(cache.OnlineDockSetKey, dockID)
	if !online {
		return nil, common.ErrDockOffline
	}

	commandID, err := tool.NextID()
	if err != nil {
		return nil, common.ErrInternal
	}
	requestID := strings.ReplaceAll(uuid.New().String(), "-", "")
	ttlMs := req.TTLMs
	if ttlMs <= 0 {
		ttlMs = 30000
	}

	paramsJSON := "{}"
	if req.Params != nil {
		if b, e := json.Marshal(req.Params); e == nil {
			paramsJSON = string(b)
		}
	}

	now := time.Now()
	cmdLog := &model.DeviceCommandLog{
		ID:          commandID,
		DockID:      dockID,
		CommandType: req.Type,
		Params:      paramsJSON,
		DroneSN:     req.DroneSN,
		RequestID:   requestID,
		TTLMs:       ttlMs,
		Status:      "sent",
		SentAt:      &now,
		CreatedAt:   now,
	}
	if err := common.DB.Create(cmdLog).Error; err != nil {
		logger.ERROR("写入指令日志失败", err)
		return nil, common.ErrInternal
	}

	if err := s.publish(cmdLog, req.Params); err != nil {
		logger.ERROR("下发指令失败", err)
		return nil, common.ErrInternal
	}
	return cmdLog, nil
}

// HandleAck 处理指令应答，按外层 requestId 更新一次发送尝试。
func (s *CommandService) HandleAck(dockID, requestID, commandIDStr string, accepted bool, resultCode string) {
	if requestID == "" {
		logger.WARN("指令应答缺少 requestId", dockID)
		return
	}

	var cmd model.DeviceCommandLog
	if err := common.DB.Where("dock_id = ? AND request_id = ? AND status = ?", dockID, requestID, "sent").First(&cmd).Error; err != nil {
		if !errors.Is(err, gorm.ErrRecordNotFound) {
			logger.ERROR("查询指令应答失败", err)
		}
		return
	}
	if commandIDStr != "" && commandIDStr != strconv.FormatInt(cmd.ID, 10) {
		logger.WARN("指令应答 commandId 与 requestId 不匹配", dockID, requestID)
		return
	}

	now := time.Now()
	updates := map[string]any{
		"ack_accepted":    accepted,
		"ack_result_code": resultCode,
		"acked_at":        now,
		"status":          "terminal",
	}
	if accepted {
		updates["status"] = "acked"
	}
	result := common.DB.Model(&model.DeviceCommandLog{}).
		Where("id = ? AND dock_id = ? AND request_id = ? AND status = ?", cmd.ID, dockID, requestID, "sent").
		Updates(updates)
	if result.Error != nil {
		logger.ERROR("更新指令应答失败", result.Error)
		return
	}
	if result.RowsAffected == 0 {
		return
	}
	if !accepted {
		failTaskExecution(cmd, resultCode)
	}
	DefaultLiveService.OnCommandAck(&cmd, accepted, resultCode)
}

func failTaskExecution(cmd model.DeviceCommandLog, reason string) {
	if reason == "" {
		reason = "COMMAND_REJECTED"
	}
	result := common.DB.Model(&model.TaskExecution{}).
		Where("dock_id = ? AND command_id = ? AND status IN ?", cmd.DockID, strconv.FormatInt(cmd.ID, 10), []string{"pending", "running"}).
		Updates(map[string]any{"status": "failed", "result_code": reason, "end_time": time.Now()})
	if result.Error != nil {
		logger.ERROR("更新任务执行失败状态失败", result.Error)
		return
	}
	if result.RowsAffected > 0 {
		DefaultTrajectoryStore.Finalize(cmd.DockID, "")
	}
}

// DispatchToDock 持久化并下发已完成权限与在线校验的设备指令。
func (s *CommandService) DispatchToDock(dockID, commandType string, params map[string]any) (*model.DeviceCommandLog, error) {
	commandID, err := tool.NextID()
	if err != nil {
		return nil, err
	}
	requestID := strings.ReplaceAll(uuid.New().String(), "-", "")
	auditParams := params
	if commandType == "video.start_stream" {
		auditParams = cloneParams(params)
		if pushURL, ok := auditParams["pushUrl"].(string); ok {
			sum := sha256.Sum256([]byte(pushURL))
			auditParams["pushUrlHash"] = hex.EncodeToString(sum[:])
			delete(auditParams, "pushUrl")
		}
	}
	auditJSON, err := json.Marshal(auditParams)
	if err != nil {
		return nil, err
	}
	now := time.Now()
	cmd := &model.DeviceCommandLog{
		ID:          commandID,
		DockID:      dockID,
		CommandType: commandType,
		Params:      string(auditJSON),
		RequestID:   requestID,
		TTLMs:       30000,
		Status:      "sent",
		SentAt:      &now,
		CreatedAt:   now,
	}
	if err := common.DB.Create(cmd).Error; err != nil {
		return nil, err
	}
	if err := s.publish(cmd, params); err != nil {
		return nil, err
	}
	return cmd, nil
}

func cloneParams(params map[string]any) map[string]any {
	copy := make(map[string]any, len(params))
	for key, value := range params {
		copy[key] = value
	}
	return copy
}

func (s *CommandService) StartRetryScanner() {
	go func() {
		ticker := time.NewTicker(10 * time.Second)
		defer ticker.Stop()
		for range ticker.C {
			s.retryTimeout()
		}
	}()
}

func (s *CommandService) retryTimeout() {
	var pending []model.DeviceCommandLog
	if err := common.DB.Where("status = 'sent'").Find(&pending).Error; err != nil {
		return
	}
	now := time.Now()
	for _, cmd := range pending {
		if cmd.SentAt == nil {
			continue
		}
		ttl := int64(cmd.TTLMs)
		if ttl <= 0 {
			ttl = 30000
		}
		if now.Sub(*cmd.SentAt).Milliseconds() <= ttl {
			continue
		}

		if cmd.RetryCount >= maxRetryCount {
			result := common.DB.Model(&model.DeviceCommandLog{}).Where("id = ? AND status = ?", cmd.ID, "sent").Update("status", "timeout")
			if result.Error == nil && result.RowsAffected > 0 {
				if cmd.CommandType == "video.start_stream" || cmd.CommandType == "video.stop_stream" {
					DefaultLiveService.OnCommandTimeout(cmd)
				} else {
					failTaskExecution(cmd, "COMMAND_ACK_TIMEOUT")
				}
			}
			continue
		}

		// 重试：复用 commandId，换新 requestId（工控机按 commandId 幂等）
		requestID := strings.ReplaceAll(uuid.New().String(), "-", "")
		_ = common.DB.Model(&model.DeviceCommandLog{}).Where("id = ?", cmd.ID).Updates(map[string]any{
			"request_id":  requestID,
			"retry_count": cmd.RetryCount + 1,
			"sent_at":     now,
		})

		var params map[string]any
		if cmd.Params != "" {
			_ = json.Unmarshal([]byte(cmd.Params), &params)
		}
		if cmd.CommandType == "video.start_stream" {
			params = DefaultLiveService.RefreshStartParams(cmd.DockID, params)
		}
		b, _ := json.Marshal(s.commandEnvelope(&cmd, requestID, params))
		topic := fmt.Sprintf("dock-edge/v1/dock/%s/command", cmd.DockID)
		_ = mqtt.Publish(topic, 1, false, b)
	}

	var acked []model.DeviceCommandLog
	if err := common.DB.Where("status = 'acked'").Find(&acked).Error; err != nil {
		return
	}
	for _, cmd := range acked {
		if cmd.AckedAt == nil {
			continue
		}
		if cmd.CommandType == "workflow.start_task" {
			s.checkWorkflowTimeout(cmd, now)
			continue
		}
		ttl := int64(cmd.TTLMs)
		if ttl <= 0 {
			ttl = 30000
		}
		if now.Sub(*cmd.AckedAt).Milliseconds() > ttl {
			failTaskExecution(cmd, "WORKFLOW_TIMEOUT")
		}
	}
}

func (s *CommandService) checkWorkflowTimeout(cmd model.DeviceCommandLog, now time.Time) {
	commandID := strconv.FormatInt(cmd.ID, 10)
	var executions []model.TaskExecution
	executionQuery := common.DB.Where("dock_id = ? AND command_id = ? AND status IN ?", cmd.DockID, commandID, []string{"pending", "running"}).Find(&executions)
	if executionQuery.Error != nil {
		logger.ERROR("查询工作流执行记录失败", executionQuery.Error)
		return
	}
	if executionQuery.RowsAffected == 0 {
		return
	}

	var workflow model.WorkflowState
	err := common.DB.Where("dock_id = ? AND command_id = ?", cmd.DockID, commandID).First(&workflow).Error
	if errors.Is(err, gorm.ErrRecordNotFound) {
		if now.Sub(*cmd.AckedAt) > workflowStartTimeout {
			s.timeoutWorkflow(cmd, "WORKFLOW_START_TIMEOUT")
		}
		return
	}
	if err != nil {
		logger.ERROR("查询工作流状态失败", err)
		return
	}
	if workflow.State == "running" && now.Sub(workflow.UpdatedAt) > workflowProgressTimeout {
		s.timeoutWorkflow(cmd, "WORKFLOW_PROGRESS_TIMEOUT")
	}
}

func (s *CommandService) timeoutWorkflow(cmd model.DeviceCommandLog, reason string) {
	result := common.DB.Model(&model.DeviceCommandLog{}).
		Where("id = ? AND status = ?", cmd.ID, "acked").
		Updates(map[string]any{"status": "terminal", "ack_result_code": reason})
	if result.Error != nil {
		logger.ERROR("更新工作流超时指令失败", result.Error)
		return
	}
	if result.RowsAffected > 0 {
		failTaskExecution(cmd, reason)
	}
}

// commandEnvelope 构造下行指令通用外层（文档 §5.1）
func (s *CommandService) commandEnvelope(cmd *model.DeviceCommandLog, requestID string, params map[string]any) *mqtt.Envelope {
	return mqtt.NewEnvelope(requestID, cmd.DockID, cmd.DroneSN, map[string]any{
		"commandId": strconv.FormatInt(cmd.ID, 10),
		"type":      cmd.CommandType,
		"ttlMs":     cmd.TTLMs,
		"params":    params,
	})
}

func (s *CommandService) publish(cmd *model.DeviceCommandLog, params map[string]any) error {
	b, err := json.Marshal(s.commandEnvelope(cmd, cmd.RequestID, params))
	if err != nil {
		return err
	}
	topic := fmt.Sprintf("dock-edge/v1/dock/%s/command", cmd.DockID)
	return mqtt.Publish(topic, 1, false, b)
}

// GetPage 指令历史分页列表（按所属机巢过滤）
func (s *CommandService) GetPage(userID int64, isAdmin bool, req *vo.CommandPageReq) (*common.PageResponse[model.DeviceCommandLog], *common.BusiError) {
	db := common.DB.Model(&model.DeviceCommandLog{}).Scopes(withDockFilter(userID, isAdmin))
	if req.Status != "" {
		db = db.Where("status = ?", req.Status)
	}
	if req.CommandType != "" {
		db = db.Where("command_type = ?", req.CommandType)
	}
	if req.DockID != "" {
		db = db.Where("dock_id = ?", req.DockID)
	}
	if req.Keyword != "" {
		kw := "%" + req.Keyword + "%"
		db = db.Where("dock_id LIKE ? OR command_type LIKE ? OR drone_sn LIKE ?", kw, kw, kw)
	}

	var total int64
	if err := db.Count(&total).Error; err != nil {
		logger.ERROR("统计指令日志失败", err)
		return nil, common.ErrInternal
	}
	var list []model.DeviceCommandLog
	if err := db.Scopes(req.Paginate).Order("sent_at DESC, id DESC").Find(&list).Error; err != nil {
		logger.ERROR("查询指令日志失败", err)
		return nil, common.ErrInternal
	}
	return common.Page(req.Pagination, total, list), nil
}

// GetDetail 指令日志详情
func (s *CommandService) GetDetail(userID int64, isAdmin bool, id int64) (*model.DeviceCommandLog, *common.BusiError) {
	var cmd model.DeviceCommandLog
	if err := common.DB.Scopes(withDockFilter(userID, isAdmin)).First(&cmd, id).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return nil, common.ErrCommandNotFound
		}
		return nil, common.ErrInternal
	}
	return &cmd, nil
}