tomswall/internal/agent/agent.go

package agent

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

	"git.unkin.net/unkin/tomswall/internal/config"
	"git.unkin.net/unkin/tomswall/internal/nftables"
)

// Applier applies a translated config to the firewall. Abstracted so the run
// loop is testable without touching the kernel.
type Applier interface {
	Apply(ctx context.Context, cfg *config.Config) error
}

// Agent runs the pull-apply-report loop for one device.
type Agent struct {
	Client   *Client
	Cache    Cache
	Interval time.Duration
	Applier  Applier
	// Resolver overrides the DNS resolver (tests); nil derives it per-config.
	Resolver *Resolver
}

// Run loops until ctx is cancelled, applying one cycle per Interval (and once
// immediately). A failed cycle is logged and retried on the next tick — the loop
// never exits on transient errors.
func (a *Agent) Run(ctx context.Context) error {
	if a.Interval <= 0 {
		a.Interval = time.Minute
	}
	t := time.NewTicker(a.Interval)
	defer t.Stop()
	for {
		if err := a.RunOnce(ctx); err != nil {
			slog.Error("agent: apply cycle failed", "err", err)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-t.C:
		}
	}
}

// RunOnce performs a single pull-apply-report cycle. On a fetch failure it falls
// back to the on-disk cache and re-applies it — it never fails closed.
func (a *Agent) RunOnce(ctx context.Context) error {
	rc, raw, err := a.Client.FetchConfig(ctx)
	if err != nil {
		slog.Warn("agent: control plane unreachable, using cached config", "err", err)
		cached, cerr := a.Cache.Read()
		if cerr != nil {
			return fmt.Errorf("read cache: %w", cerr)
		}
		if cached == nil {
			return fmt.Errorf("control plane unreachable and no cached config: %w", err)
		}
		// Re-apply last known-good; do not report a generation we didn't fetch.
		return a.applyConfig(ctx, cached, false)
	}

	if err := a.Cache.Write(raw); err != nil {
		slog.Warn("agent: caching config failed", "err", err)
	}
	return a.applyConfig(ctx, rc, true)
}

func (a *Agent) applyConfig(ctx context.Context, rc *RenderedConfig, report bool) error {
	resolver := a.Resolver
	if resolver == nil {
		resolver = NewResolver(rc.Resolver)
	}
	resolver.ExpandDNSSets(ctx, rc)

	cfg, err := Translate(rc)
	if err != nil {
		return fmt.Errorf("translate: %w", err)
	}
	if err := a.Applier.Apply(ctx, cfg); err != nil {
		return fmt.Errorf("apply: %w", err)
	}
	slog.Info("agent: applied config", "generation", rc.Generation, "rules", len(cfg.Rules))

	if report {
		if err := a.Client.ReportStatus(ctx, rc.Generation); err != nil {
			slog.Warn("agent: reporting status failed", "err", err)
		}
		// Report the FIB so the control plane can scope router enforcement.
		if fib := CollectFIB(ctx); len(fib) > 0 {
			if err := a.Client.ReportRoutes(ctx, fib); err != nil {
				slog.Warn("agent: reporting routes failed", "err", err)
			}
		}
	}
	return nil
}

// EngineApplier applies via the real nftables differential engine.
type EngineApplier struct{}

// Apply computes and applies the differential change set for cfg.
func (EngineApplier) Apply(_ context.Context, cfg *config.Config) error {
	engine, err := nftables.NewEngine(cfg)
	if err != nil {
		return fmt.Errorf("initializing nftables: %w", err)
	}
	changes, err := engine.Plan()
	if err != nil {
		return fmt.Errorf("computing changes: %w", err)
	}
	if changes.Empty() {
		return nil
	}
	return engine.Apply(changes)
}