felhom-agent/internal/pbsdr/loop.go

package pbsdr

import (
	"context"
	"log/slog"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
)

// Loop drives the Manager on its own cadence AND consumes fetched desired-state via the
// desired.Syncer raw-consumer seam (the wgtunnel Loop shape). fetched=false ("no desired data
// seen yet") is never a signal; an absent pbs_dr block on a PRESENT desired-state is a plain
// no-op this slice (no teardown — deprovision is a deliberate future op).
type Loop struct {
	mgr      *Manager
	interval time.Duration
	logger   *slog.Logger

	mu      sync.Mutex
	fetched bool
	block   *hub.WirePBSDR

	nudge chan struct{}
}

// NewLoop builds the loop. interval defaults to 60s.
func NewLoop(mgr *Manager, interval time.Duration, logger *slog.Logger) *Loop {
	if interval <= 0 {
		interval = 60 * time.Second
	}
	if logger == nil {
		logger = slog.Default()
	}
	return &Loop{mgr: mgr, interval: interval, logger: logger, nudge: make(chan struct{}, 1)}
}

// OnDesiredState implements desired.RawConsumer: store the latest pbs_dr block (or its absence)
// and nudge the loop. Non-blocking and panic-free by construction.
func (l *Loop) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse) {
	if resp == nil {
		return
	}
	l.mu.Lock()
	l.fetched = true
	l.block = resp.DesiredState.PBSDR
	l.mu.Unlock()
	select {
	case l.nudge <- struct{}{}:
	default:
	}
}

func (l *Loop) snapshot() (bool, *hub.WirePBSDR) {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.fetched, l.block
}

// Run applies immediately, then on every tick or desired-state nudge, until ctx is cancelled.
func (l *Loop) Run(ctx context.Context) error {
	fetched, block := l.snapshot()
	l.mgr.Apply(ctx, fetched, block)
	t := time.NewTicker(l.interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-t.C:
		case <-l.nudge:
		}
		fetched, block = l.snapshot()
		l.mgr.Apply(ctx, fetched, block)
	}
}

// PBSDRStatus implements the hub collector's PBSDRReporter seam.
func (l *Loop) PBSDRStatus(_ context.Context) *hub.PBSDRStatus {
	return l.mgr.Status()
}