EvoBGP/internal/httpapi/speakers_live.go

package httpapi

import (
	"context"
	"strings"
	"sync"
	"time"

	"evobgp/internal/birdfmt"
	"evobgp/internal/nodedispatch"
	"evobgp/internal/store"
)

func countBGPSessions(sessions []birdfmt.BGPSession) (total, established int) {
	total = len(sessions)
	for _, s := range sessions {
		if strings.EqualFold(strings.TrimSpace(s.State), "Established") {
			established++
		}
	}
	return total, established
}

func speakerLiveStatusJSON(sp *store.Speaker, view speakerBGPLive, health *nodedispatch.AgentHealthResult) map[string]any {
	total, established := countBGPSessions(view.Sessions)
	m := map[string]any{
		"label":              view.Label,
		"bgp_poll_ok":        view.Error == "",
		"bgp_sessions_total": total,
		"bgp_established":    established,
	}
	if view.Error != "" {
		m["bgp_poll_error"] = view.Error
	}
	if health != nil {
		m["agent_ok"] = health.OK
		if health.Error != "" {
			m["agent_error"] = health.Error
		}
		if health.LastSyncAt != "" {
			m["agent_last_sync_at"] = health.LastSyncAt
		}
		if health.LastAppliedRevisionID != "" {
			m["agent_last_applied_revision_id"] = health.LastAppliedRevisionID
		}
	} else if sp != nil && strings.EqualFold(strings.TrimSpace(sp.Role), "master") {
		m["agent_ok"] = view.Error == ""
		if view.Error != "" {
			m["agent_error"] = view.Error
		}
	} else if sp != nil && store.SpeakerNeedsRemoteDispatch(sp.Role, store.ParseSpeakerMeta(sp.MetaJSON)) {
		m["agent_ok"] = false
		m["agent_error"] = "agent health not polled"
	}
	if len(view.Sessions) > 0 {
		sess := make([]map[string]any, 0, len(view.Sessions))
		for _, s := range view.Sessions {
			row := map[string]any{
				"name":  s.Name,
				"state": s.State,
			}
			if strings.TrimSpace(s.Neighbor) != "" {
				row["neighbor"] = s.Neighbor
			}
			sess = append(sess, row)
		}
		m["sessions"] = sess
	}
	return m
}

func (s *Server) collectSpeakerLiveStatus(ctx context.Context, tenantID string, fresh bool, speakers []*store.Speaker) map[string]map[string]any {
	views := s.collectSpeakerBGPLive(ctx, tenantID, fresh)
	viewByID := make(map[string]speakerBGPLive, len(views))
	for _, v := range views {
		if v.SpeakerID != "" {
			viewByID[v.SpeakerID] = v
		}
	}

	opts := nodedispatch.Options{Timeout: 8 * time.Second}
	type healthWrap struct {
		id string
		h  nodedispatch.AgentHealthResult
	}
	healthCh := make(chan healthWrap, len(speakers))
	var wg sync.WaitGroup
	for _, sp := range speakers {
		if sp == nil {
			continue
		}
		meta := store.ParseSpeakerMeta(sp.MetaJSON)
		if !store.SpeakerNeedsRemoteDispatch(sp.Role, meta) {
			continue
		}
		wg.Add(1)
		go func(speaker *store.Speaker) {
			defer wg.Done()
			healthCh <- healthWrap{
				id: speaker.ID,
				h:  nodedispatch.FetchAgentHealth(ctx, speaker, opts),
			}
		}(sp)
	}
	wg.Wait()
	close(healthCh)
	healthByID := make(map[string]nodedispatch.AgentHealthResult, len(speakers))
	for hw := range healthCh {
		healthByID[hw.id] = hw.h
	}

	out := make(map[string]map[string]any, len(speakers))
	for _, sp := range speakers {
		if sp == nil {
			continue
		}
		view, ok := viewByID[sp.ID]
		if !ok {
			view = speakerBGPLive{SpeakerID: sp.ID, Label: speakerDisplayLabel(sp)}
		}
		var hp *nodedispatch.AgentHealthResult
		if h, ok := healthByID[sp.ID]; ok {
			hCopy := h
			hp = &hCopy
		}
		out[sp.ID] = speakerLiveStatusJSON(sp, view, hp)
	}
	return out
}