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) } } 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) }