iop/apps/control-plane/cmd/control-plane/fleet_orchestrator.go

package main

import (
	"sync"
	"time"

	"iop/apps/control-plane/internal/wire"
)

// fleetOptions configures the fleet orchestration service: per-call timeouts,
// bounded concurrency for status/command fan-out, and a short freshness cache
// for connected-edge status views. Zero values fall back to package defaults so
// callers and tests can override only the fields they care about.
type fleetOptions struct {
	StatusTimeout         time.Duration
	CommandTimeout        time.Duration
	MaxConcurrentStatus   int
	MaxConcurrentCommands int
	StatusCacheTTL        time.Duration
	Now                   func() time.Time
}

const (
	defaultFleetStatusTimeout         = 5 * time.Second
	defaultFleetCommandTimeout        = 10 * time.Second
	defaultFleetMaxConcurrentStatus   = 4
	defaultFleetMaxConcurrentCommands = 4
	defaultFleetStatusCacheTTL        = 2 * time.Second
)

func (o fleetOptions) withDefaults() fleetOptions {
	if o.StatusTimeout <= 0 {
		o.StatusTimeout = defaultFleetStatusTimeout
	}
	if o.CommandTimeout <= 0 {
		o.CommandTimeout = defaultFleetCommandTimeout
	}
	if o.MaxConcurrentStatus <= 0 {
		o.MaxConcurrentStatus = defaultFleetMaxConcurrentStatus
	}
	if o.MaxConcurrentCommands <= 0 {
		o.MaxConcurrentCommands = defaultFleetMaxConcurrentCommands
	}
	if o.StatusCacheTTL <= 0 {
		o.StatusCacheTTL = defaultFleetStatusCacheTTL
	}
	if o.Now == nil {
		o.Now = time.Now
	}
	return o
}

// fleetService orchestrates fleet-wide status and command fan-out over the Edge
// registry with bounded concurrency, per-call timeouts, and a short freshness
// cache for connected-edge status. It keeps the existing JSON contracts; it only
// changes how the views are gathered.
type fleetService struct {
	registry      *wire.EdgeRegistry
	requestStatus edgeStatusRequester
	sendCommand   edgeCommandSender
	opts          fleetOptions

	cacheMu sync.Mutex
	cache   map[string]fleetStatusCacheEntry
}

type fleetStatusCacheEntry struct {
	view   fleetEdgeView
	expiry time.Time
}

func newFleetService(registry *wire.EdgeRegistry, requestStatus edgeStatusRequester, sendCommand edgeCommandSender, opts fleetOptions) *fleetService {
	return &fleetService{
		registry:      registry,
		requestStatus: requestStatus,
		sendCommand:   sendCommand,
		opts:          opts.withDefaults(),
		cache:         make(map[string]fleetStatusCacheEntry),
	}
}

// Status returns one view per registered Edge in registry snapshot order.
// Connected-edge status is fetched with bounded concurrency and served from a
// short freshness cache; disconnected edges reflect registry state immediately
// and are never cached.
func (s *fleetService) Status() fleetStatusResponse {
	states := s.registry.Snapshots()
	views := make([]fleetEdgeView, len(states))

	sem := make(chan struct{}, s.opts.MaxConcurrentStatus)
	var wg sync.WaitGroup
	for i, state := range states {
		// Disconnected edges (and the no-requester case) need no wire call and
		// must reflect registry state immediately, so they bypass the cache and
		// the concurrency-bounded worker pool.
		if !state.Connected || s.requestStatus == nil {
			views[i] = fleetEdgeViewFromState(state, s.requestStatus, s.opts.StatusTimeout)
			continue
		}
		if cached, ok := s.cachedStatus(state.EdgeID); ok {
			views[i] = cached
			continue
		}
		wg.Add(1)
		sem <- struct{}{}
		go func(i int, state wire.EdgeConnectionState) {
			defer wg.Done()
			defer func() { <-sem }()
			view := fleetEdgeViewFromState(state, s.requestStatus, s.opts.StatusTimeout)
			s.storeStatus(state.EdgeID, view)
			views[i] = view
		}(i, state)
	}
	wg.Wait()

	return fleetStatusResponse{GeneratedAt: s.opts.Now().UTC(), Edges: views}
}

func (s *fleetService) cachedStatus(edgeID string) (fleetEdgeView, bool) {
	s.cacheMu.Lock()
	defer s.cacheMu.Unlock()
	entry, ok := s.cache[edgeID]
	if !ok || !entry.expiry.After(s.opts.Now()) {
		return fleetEdgeView{}, false
	}
	return entry.view, true
}

func (s *fleetService) storeStatus(edgeID string, view fleetEdgeView) {
	s.cacheMu.Lock()
	defer s.cacheMu.Unlock()
	s.cache[edgeID] = fleetStatusCacheEntry{
		view:   view,
		expiry: s.opts.Now().Add(s.opts.StatusCacheTTL),
	}
}

// Command fans out a surface-neutral command to every connected Edge with
// bounded concurrency, preserving connected snapshot order in the results.
// Disconnected edges are skipped and per-edge errors are surfaced in the result
// view. The caller is responsible for body validation and the nil-dispatcher
// case before invoking Command.
func (s *fleetService) Command(operation, targetSelector string, parameters map[string]string) fleetCommandResponse {
	connected := make([]wire.EdgeConnectionState, 0)
	for _, state := range s.registry.Snapshots() {
		if state.Connected {
			connected = append(connected, state)
		}
	}

	results := make([]edgeCommandResponseView, len(connected))
	sem := make(chan struct{}, s.opts.MaxConcurrentCommands)
	var wg sync.WaitGroup
	for i, state := range connected {
		wg.Add(1)
		sem <- struct{}{}
		go func(i int, state wire.EdgeConnectionState) {
			defer wg.Done()
			defer func() { <-sem }()
			resp, err := s.sendCommand(state.EdgeID, operation, targetSelector, parameters, s.opts.CommandTimeout)
			if err != nil {
				results[i] = edgeCommandResponseView{
					EdgeID: state.EdgeID,
					Status: "error",
					Error:  err.Error(),
				}
				return
			}
			results[i] = edgeCommandResponseViewFromProto(resp)
		}(i, state)
	}
	wg.Wait()

	return fleetCommandResponse{Operation: operation, Results: results}
}