EvoBGP/internal/httpapi/peers_live.go

package httpapi

import (
	"context"
	"net/netip"
	"os"
	"strings"
	"sync"
	"time"

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

const peerLiveCacheTTL = 15 * time.Second

type speakerBGPLive struct {
	SpeakerID string
	Label     string
	Sessions  []birdfmt.BGPSession
	Error     string
}

type peerLiveCacheEntry struct {
	at    time.Time
	views []speakerBGPLive
}

var peerLiveCache sync.Map // tenantID -> peerLiveCacheEntry

type peerSessionOnSpeaker struct {
	SpeakerID string `json:"speaker_id"`
	Label     string `json:"label"`
	State     string `json:"state"`
	PollError string `json:"poll_error,omitempty"`
}

type liveSpeakerPoll struct {
	SpeakerID    string `json:"speaker_id"`
	Label        string `json:"label"`
	OK           bool   `json:"ok"`
	SessionCount int    `json:"session_count"`
	PollError    string `json:"poll_error,omitempty"`
}

func speakerDisplayLabel(sp *store.Speaker) string {
	if sp == nil {
		return ""
	}
	meta := store.ParseSpeakerMeta(sp.MetaJSON)
	host := strings.TrimSpace(meta.AgentDomain)
	if host == "" {
		host = strings.TrimSpace(sp.Endpoint)
	}
	if strings.EqualFold(strings.TrimSpace(sp.Role), "master") {
		if host != "" {
			return "CP · " + host
		}
		return "CP (master)"
	}
	if host != "" {
		return host
	}
	return sp.ID
}

func masterSpeakerID(speakers []*store.Speaker) string {
	for _, sp := range speakers {
		if sp != nil && strings.EqualFold(strings.TrimSpace(sp.Role), "master") {
			return sp.ID
		}
	}
	return ""
}

func (s *Server) collectSpeakerBGPLive(ctx context.Context, tenantID string, fresh bool) []speakerBGPLive {
	if !fresh {
		if v, ok := peerLiveCache.Load(tenantID); ok {
			ent := v.(peerLiveCacheEntry)
			if time.Since(ent.at) < peerLiveCacheTTL {
				return ent.views
			}
		}
	}
	speakers := s.store.ListSpeakersForTenant(tenantID)
	views := make([]speakerBGPLive, 0, len(speakers)+1)

	if sock := strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_SOCKET")); sock != "" {
		v := speakerBGPLive{Label: "CP (local BIRD)"}
		if mid := masterSpeakerID(speakers); mid != "" {
			v.SpeakerID = mid
			for _, sp := range speakers {
				if sp != nil && sp.ID == mid {
					v.Label = speakerDisplayLabel(sp)
					break
				}
			}
		}
		out, err := birdfmt.ShowProtocols(ctx, sock, strings.TrimSpace(os.Getenv("EVOBGP_BIRDC_BIN")))
		if err != nil {
			v.Error = err.Error()
		} else {
			v.Sessions = birdfmt.ParseBGPSessions(out)
		}
		views = append(views, v)
	}

	opts := nodedispatch.Options{Timeout: 8 * time.Second}
	type resWrap struct {
		sp  *store.Speaker
		res nodedispatch.BirdProtocolsResult
	}
	ch := make(chan resWrap, 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()
			ch <- resWrap{
				sp:  speaker,
				res: nodedispatch.FetchBirdProtocols(ctx, speaker, opts),
			}
		}(sp)
	}
	wg.Wait()
	close(ch)
	for rw := range ch {
		views = append(views, speakerBGPLive{
			SpeakerID: rw.sp.ID,
			Label:     speakerDisplayLabel(rw.sp),
			Sessions:  rw.res.Sessions,
			Error:     rw.res.Error,
		})
	}

	peerLiveCache.Store(tenantID, peerLiveCacheEntry{at: time.Now(), views: views})
	return views
}

func liveSpeakerPollJSON(views []speakerBGPLive) []liveSpeakerPoll {
	out := make([]liveSpeakerPoll, 0, len(views))
	for _, v := range views {
		out = append(out, liveSpeakerPoll{
			SpeakerID:    v.SpeakerID,
			Label:        v.Label,
			OK:           v.Error == "",
			SessionCount: len(v.Sessions),
			PollError:    v.Error,
		})
	}
	return out
}

func findPeerSession(sessions []birdfmt.BGPSession, protoName string, neighbor netip.Addr, hasNeighbor bool) *birdfmt.BGPSession {
	for i := range sessions {
		if peerSessionMatches(sessions[i], protoName, neighbor, hasNeighbor) {
			return &sessions[i]
		}
	}
	return nil
}

func matchPeerOnSpeakers(peer *store.BGPPeer, views []speakerBGPLive) (
	bestState string,
	connectedID string,
	connectedLabel string,
	establishedOn []peerSessionOnSpeaker,
	on []peerSessionOnSpeaker,
	mismatch bool,
) {
	if peer == nil {
		return "", "", "", nil, nil, false
	}
	neighbor, hasNeighbor := store.ParsePeerNeighbor(peer.Neighbor)
	protoName := birdfmt.PeerProtocolName(peer.ID)

	for _, v := range views {
		if v.Error != "" && len(v.Sessions) == 0 {
			on = append(on, peerSessionOnSpeaker{
				SpeakerID: v.SpeakerID,
				Label:     v.Label,
				PollError: v.Error,
			})
			continue
		}
		sess := findPeerSession(v.Sessions, protoName, neighbor, hasNeighbor)
		if sess == nil {
			on = append(on, peerSessionOnSpeaker{
				SpeakerID: v.SpeakerID,
				Label:     v.Label,
				State:     "absent",
			})
			continue
		}
		hit := peerSessionOnSpeaker{
			SpeakerID: v.SpeakerID,
			Label:     v.Label,
			State:     sess.State,
		}
		on = append(on, hit)
		if strings.EqualFold(strings.TrimSpace(sess.State), "Established") {
			establishedOn = append(establishedOn, hit)
		}
		if bestState == "" || sessionStateRank(sess.State) > sessionStateRank(bestState) {
			bestState = sess.State
		}
	}

	if len(establishedOn) > 0 {
		bestState = "Established"
		labels := make([]string, 0, len(establishedOn))
		for _, e := range establishedOn {
			labels = append(labels, e.Label)
		}
		connectedLabel = strings.Join(labels, ", ")
		if len(establishedOn) == 1 {
			connectedID = establishedOn[0].SpeakerID
		}
	} else if len(on) == 1 && on[0].PollError == "" && on[0].State != "" {
		connectedID = on[0].SpeakerID
		connectedLabel = on[0].Label
	}

	if peer.SpeakerID != nil && strings.TrimSpace(*peer.SpeakerID) != "" && len(establishedOn) > 0 {
		want := strings.TrimSpace(*peer.SpeakerID)
		found := false
		for _, e := range establishedOn {
			if strings.EqualFold(strings.TrimSpace(e.SpeakerID), want) {
				found = true
				break
			}
		}
		mismatch = !found
	}
	return bestState, connectedID, connectedLabel, establishedOn, on, mismatch
}

func peerSessionMatches(sess birdfmt.BGPSession, protoName string, neighbor netip.Addr, hasNeighbor bool) bool {
	if strings.EqualFold(strings.TrimSpace(sess.Name), protoName) {
		return true
	}
	if !hasNeighbor || strings.TrimSpace(sess.Neighbor) == "" {
		return false
	}
	peerAddr, ok := store.ParsePeerNeighbor(sess.Neighbor)
	if !ok {
		return false
	}
	return peerAddr == neighbor
}

func sessionStateRank(state string) int {
	switch strings.ToLower(strings.TrimSpace(state)) {
	case "established":
		return 100
	case "openconfirm", "opensent":
		return 80
	case "active", "connect":
		return 60
	case "idle":
		return 20
	default:
		return 10
	}
}

func applyPeerLiveFields(row map[string]any, peer *store.BGPPeer, views []speakerBGPLive) {
	state, connID, connLabel, establishedOn, on, mismatch := matchPeerOnSpeakers(peer, views)
	row["session_on_speakers"] = on
	row["established_on_speakers"] = establishedOn
	row["session_conflict"] = false
	row["session_mismatch"] = mismatch
	if state != "" {
		row["session_state"] = state
	}
	if connLabel != "" {
		row["connected_speaker_label"] = connLabel
	}
	row["connected_speaker_id"] = peerLiveSpeakerIDOrNull(connID)
}

func peerLiveSpeakerIDOrNull(id string) any {
	if strings.TrimSpace(id) == "" {
		return nil
	}
	return id
}