yovision/Sense/server/app/sense/edge_node/service.go

package edge_node

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	coreService "github.com/go-admin-team/go-admin-core/sdk/service"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

var (
	ErrInvalidQuery  = errors.New("边缘节点查询条件不符合要求")
	ErrNotFound      = errors.New("边缘节点不存在")
	ErrInvalidSample = errors.New("边缘节点心跳样本不符合要求")
)

type Service struct {
	coreService.Service
	Now func() time.Time
}

func NewService(db *gorm.DB) *Service {
	return &Service{Service: coreService.Service{Orm: db}, Now: time.Now}
}

func (s *Service) List(request PageRequest) (PageResponse, error) {
	if request.GetPageSize() > 100 {
		return PageResponse{}, ErrInvalidQuery
	}
	var count int64
	if err := s.Orm.Model(&Node{}).Count(&count).Error; err != nil {
		return PageResponse{}, fmt.Errorf("count edge nodes: %w", err)
	}
	var models []Node
	if err := s.Orm.Order("name ASC, id ASC").Limit(request.GetPageSize()).Offset((request.GetPageIndex() - 1) * request.GetPageSize()).Find(&models).Error; err != nil {
		return PageResponse{}, fmt.Errorf("list edge nodes: %w", err)
	}
	result := PageResponse{List: make([]Response, 0, len(models)), Count: count}
	result.Summary.HeartbeatTimeout = int64(HeartbeatTimeout / time.Second)
	var all []Node
	if err := s.Orm.Find(&all).Error; err != nil {
		return PageResponse{}, fmt.Errorf("summarize edge nodes: %w", err)
	}
	result.Summary.Total = len(all)
	for _, item := range all {
		status := s.status(item)
		switch status {
		case StatusOffline:
			result.Summary.Offline++
		case StatusRecovering:
			result.Summary.Recovering++
		default:
			result.Summary.Online++
		}
		result.Summary.LoadUsed += item.LoadUsed
		result.Summary.LoadCapacity += item.LoadCapacity
		result.Summary.BackfillQueueDepth += item.BackfillQueueDepth
	}
	for _, item := range models {
		result.List = append(result.List, s.response(item, nil))
	}
	return result, nil
}

func (s *Service) Get(id string) (Response, error) {
	id = strings.TrimSpace(id)
	if id == "" || len(id) > 64 {
		return Response{}, ErrInvalidQuery
	}
	var model Node
	if err := s.Orm.First(&model, "id = ?", id).Error; err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			return Response{}, ErrNotFound
		}
		return Response{}, fmt.Errorf("get edge node: %w", err)
	}
	var events []Event
	if err := s.Orm.Where("node_id = ?", id).Order("occurred_at DESC, id DESC").Limit(20).Find(&events).Error; err != nil {
		return Response{}, fmt.Errorf("list edge node events: %w", err)
	}
	return s.response(model, events), nil
}

// ApplyHeartbeat updates the Sense-owned projection atomically and records
// only state transitions. Callers are internal adapters, not public APIs.
func (s *Service) ApplyHeartbeat(ctx context.Context, sample HeartbeatSample) error {
	now := sample.CollectedAt.UTC()
	if now.IsZero() {
		now = s.now()
	}
	sample.ID = strings.TrimSpace(sample.ID)
	if sample.ID == "" || len(sample.ID) > 64 || strings.TrimSpace(sample.Name) == "" || sample.LastHeartbeatAt.IsZero() || sample.StartedAt.IsZero() || sample.LoadUsed < 0 || sample.LoadCapacity < 0 || sample.LoadUsed > sample.LoadCapacity || sample.BackfillQueueDepth < 0 {
		return ErrInvalidSample
	}
	sample.LastHeartbeatAt = sample.LastHeartbeatAt.UTC()
	sample.LastCollectedAt = now
	return s.Orm.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		var previous Node
		err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&previous, "id = ?", sample.ID).Error
		if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
			return fmt.Errorf("lock edge node projection: %w", err)
		}
		from := ""
		if err == nil {
			from = statusAt(previous, now)
			sample.CreatedAt = previous.CreatedAt
			sample.ProjectionVersion = previous.ProjectionVersion + 1
		} else {
			sample.ProjectionVersion = 1
		}
		if err = tx.Save(&sample.Node).Error; err != nil {
			return fmt.Errorf("save edge node projection: %w", err)
		}
		to := statusAt(sample.Node, now)
		if from == to && from != "" {
			return nil
		}
		if from == StatusOffline {
			timeoutAt := previous.LastHeartbeatAt.UTC().Add(HeartbeatTimeout)
			if err = tx.Create(&Event{NodeID: sample.ID, EventType: "heartbeat_timeout", FromStatus: StatusOnline, ToStatus: StatusOffline, Detail: "节点心跳超过 90 秒未更新", OccurredAt: timeoutAt}).Error; err != nil {
				return err
			}
		}
		eventType, detail := "heartbeat_received", "收到节点首次心跳"
		if from == StatusOffline {
			eventType, detail = "node_recovered", "节点心跳恢复，通道状态继续独立收敛"
		} else if to == StatusRecovering {
			eventType, detail = "recovery_started", "节点进入恢复阶段"
		}
		return tx.Create(&Event{NodeID: sample.ID, EventType: eventType, FromStatus: from, ToStatus: to, Detail: detail, OccurredAt: now}).Error
	})
}

func (s *Service) response(model Node, events []Event) Response {
	now := s.now()
	freshness := int64(now.Sub(model.LastHeartbeatAt.UTC()).Seconds())
	if freshness < 0 {
		freshness = 0
	}
	uptime := int64(model.LastHeartbeatAt.UTC().Sub(model.StartedAt.UTC()).Seconds())
	if uptime < 0 {
		uptime = 0
	}
	return Response{ID: model.ID, Name: model.Name, Location: model.Location, Status: statusAt(model, now), RuntimeVersion: model.RuntimeVersion, ModelVersion: model.ModelVersion, StartedAt: model.StartedAt, UptimeSeconds: uptime, LastHeartbeatAt: model.LastHeartbeatAt, LastCollectedAt: model.LastCollectedAt, FreshnessSeconds: freshness, Stale: freshness > int64(HeartbeatTimeout/time.Second), LoadUsed: model.LoadUsed, LoadCapacity: model.LoadCapacity, ControlTunnelStatus: model.ControlTunnelStatus, ControlTunnelDetail: model.ControlTunnelDetail, ControlLatencyMs: model.ControlLatencyMs, VideoPlaneStatus: model.VideoPlaneStatus, VideoPlaneDetail: model.VideoPlaneDetail, ActiveStreamCount: model.ActiveStreamCount, BackfillStatus: model.BackfillStatus, BackfillQueueDepth: model.BackfillQueueDepth, RecoveryPhase: model.RecoveryPhase, ProjectionVersion: model.ProjectionVersion, Events: events}
}

func (s *Service) status(model Node) string { return statusAt(model, s.now()) }
func (s *Service) now() time.Time {
	if s.Now != nil {
		return s.Now().UTC()
	}
	return time.Now().UTC()
}

func statusAt(model Node, now time.Time) string {
	if now.UTC().Sub(model.LastHeartbeatAt.UTC()) > HeartbeatTimeout {
		return StatusOffline
	}
	if strings.TrimSpace(model.RecoveryPhase) != "" && model.RecoveryPhase != "converged" {
		return StatusRecovering
	}
	return StatusOnline
}