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 }