felhom-agent/internal/felhomsshd/loop.go

package felhomsshd

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

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

// Loop drives the felhom-sshd Manager on its own cadence (the wgtunnel/lanresolver shape) and
// consumes the hub desired-state's wireguard block via the desired.Syncer raw-consumer seam (for
// oob_peer_ip → the belt, and oob_operator_ssh_key → authorized_keys). Each tick: claim/render/reload
// the instance, sync the belt sets, and run the health/heal check.
type Loop struct {
	mgr      *Manager
	belt     *Belt // Part 3 (nil-safe: no belt sync when unset)
	interval time.Duration
	logger   *slog.Logger

	mu      sync.Mutex
	fetched bool // a desired-state document has been delivered (false = never a teardown signal)
	block   *hub.WireWireguard

	nudge chan struct{}
}

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

// OnDesiredState implements desired.RawConsumer: store the latest wireguard block and nudge.
func (l *Loop) OnDesiredState(_ context.Context, resp *hub.DesiredStateResponse) {
	if resp == nil {
		return
	}
	l.mu.Lock()
	l.fetched = true
	l.block = resp.DesiredState.Wireguard
	l.mu.Unlock()
	select {
	case l.nudge <- struct{}{}:
	default:
	}
}

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

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

// reconcile runs one full pass: instance apply → belt sync → health/heal. The `fetched` gate mirrors
// wgtunnel: until a desired-state document arrives, the operator inputs (authorized_keys, @operator_ips)
// are LEFT UNTOUCHED — an unfetched block must never empty the belt (operator lockout) or wipe the key.
func (l *Loop) reconcile(ctx context.Context) {
	fetched, block := l.snapshot()
	// Pass the block to Apply only when fetched, so authorized_keys is applied only from real desired
	// state (a nil block on Apply skips the authorized_keys write, leaving the existing file).
	applyBlock := block
	if !fetched {
		applyBlock = nil
	}
	port, err := l.mgr.Apply(ctx, applyBlock)
	if err != nil {
		return // Apply logged; a claim/exhaustion or install error — retry next tick
	}
	if l.belt != nil {
		l.belt.SyncPort(ctx, port) // always — the port is the agent's own claim
		if fetched {
			l.belt.SyncOperator(ctx, oobPeerIP(block)) // only from real desired state
		}
	}
	l.mgr.HealAndCheck(ctx, port)
}

// oobPeerIP extracts the operator /32 source (bare IP) from the block, or "" when OOB is off.
func oobPeerIP(block *hub.WireWireguard) string {
	if block == nil {
		return ""
	}
	return block.OOBPeerIP
}

// OOBStatus implements the hub collector's reporter seam (Part 4).
func (l *Loop) OOBStatus(ctx context.Context) *hub.OOBStatus {
	_, block := l.snapshot()
	return l.mgr.Status(ctx, block)
}