fancy-dns/internal/app/app.go

// Package app wires the components together and owns the process lifecycle:
// configuration, listeners, background loops, the admin API and shutdown.
package app

import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/http/pprof"
	"net/netip"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"git.k8s-git-k6n.net/mats/fancy-dns/internal/accesslog"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/acl"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/api"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/blocklist"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/cache"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/cluster"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/config"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/engine"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/kubesource"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/kubezones"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/localzone"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/metrics"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/resolver"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/server"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/web"
	"github.com/miekg/dns"
)

// Version is the build version, overridden at link time with -ldflags.
var Version = "dev"

// Options configures an App.
type Options struct {
	// ConfigPath is the configuration file; "" runs on defaults.
	ConfigPath string
	// Version is the reported build version.
	Version string
	// LogLevel and LogFormat override the config's logging settings.
	LogLevel  string
	LogFormat string
	// Kube turns on the Kubernetes source for groups, lists and zones.
	Kube KubeOptions
	// AccessLogOut receives the access log; nil means os.Stdout.
	AccessLogOut io.Writer
}

// App is the running process.
type App struct {
	opt Options
	log *slog.Logger
	cfg *config.Store
	ver string

	blocks     *blocklist.Manager
	overrideMu sync.Mutex
	cache      *cache.Manager
	cstore     *cache.Store
	logger     *accesslog.Logger
	aclH       *acl.Holder
	local      *localzone.Holder
	// zones is the Kubernetes-sourced zone client.
	zones *kubezones.Client
	// kube watches the cluster's fancy-dns objects when enabled; kubeSnap is
	// what the config overlay merges in.
	kube     *kubesource.Source
	kubeSnap atomic.Pointer[kubesource.Snapshot]
	// managedZones is the console-managed zone directory: the only zone data
	// the admin API may rewrite. An empty directory is what makes the
	// console's Zones page read-only.
	managedZones *localzone.Managed
	pool         *resolver.Pool
	reverse      *engine.ReverseCache
	engine       *engine.Engine
	dns          *server.Server
	api          *api.Server
	cl           *cluster.Cluster

	apiServers []*http.Server
	apiLn      []net.Listener

	// reloadMu serialises reload work, since SIGHUP, the API and the cluster
	// can all trigger one at the same time.
	reloadMu sync.Mutex
	// zoneReloadMu serialises file-watcher and config-swap recompiles.
	zoneReloadMu sync.Mutex

	startedAt time.Time
	// degraded is set when a background component failed; surfaced in health.
	degradedMu sync.RWMutex
	degraded   []string
}

// New builds an App from options. It does not bind anything yet; call Start.
func New(opt Options) (*App, error) {
	if opt.Version == "" {
		opt.Version = Version
	}
	cfg, err := config.Load(opt.ConfigPath)
	if err != nil {
		return nil, err
	}
	log, err := newLogger(cfg, opt)
	if err != nil {
		return nil, err
	}
	store := config.NewStore(cfg, opt.ConfigPath)
	zones, err := kubezones.InCluster()
	if err != nil {
		return nil, err
	}
	return &App{opt: opt, log: log, cfg: store, ver: opt.Version, zones: zones}, nil
}

// newLogger builds the structured logger from config plus CLI overrides.
func newLogger(cfg *config.Config, opt Options) (*slog.Logger, error) {
	level := cfg.Log.Level
	if opt.LogLevel != "" {
		level = opt.LogLevel
	}
	var lvl slog.Level
	switch level {
	case "debug":
		lvl = slog.LevelDebug
	case "warn":
		lvl = slog.LevelWarn
	case "error":
		lvl = slog.LevelError
	default:
		lvl = slog.LevelInfo
	}
	format := cfg.Log.Format
	if opt.LogFormat != "" {
		format = opt.LogFormat
	}
	handlerOpts := &slog.HandlerOptions{Level: lvl}

	var w *os.File = os.Stderr
	if cfg.Log.Addr != "" {
		if strings.HasPrefix(cfg.Log.Addr, "/") {
			f, err := os.OpenFile(cfg.Log.Addr, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
			if err != nil {
				return nil, fmt.Errorf("open log file: %w", err)
			}
			w = f
		}
	}
	var h slog.Handler
	if format == "json" {
		h = slog.NewJSONHandler(w, handlerOpts)
	} else {
		h = slog.NewTextHandler(w, handlerOpts)
	}
	return slog.New(h).With("service", "fancy-dns", "version", opt.Version), nil
}

// Start brings up every component in dependency order.
func (a *App) Start(ctx context.Context) error {
	a.startedAt = time.Now()
	cfg := a.cfg.Get()
	metrics.SetUp(true)

	if err := a.startCache(cfg); err != nil {
		return err
	}
	if err := a.startLogger(cfg); err != nil {
		return err
	}
	if err := a.startKube(ctx); err != nil {
		return err
	}
	cfg = a.cfg.Get() // the overlay may have changed it
	a.startBlocklist()
	a.startLocalZones(cfg)
	a.startUpstream(cfg)
	a.startDNS(ctx)
	if err := a.startAPI(ctx); err != nil {
		return err
	}
	a.startCluster(ctx, cfg)
	a.startBackground(ctx, cfg)
	a.cfg.OnSwap(a.onConfigSwap)

	// The real bound admin address, not the configured one: with :0 the config
	// is useless for telling the operator where to point a browser.
	a.log.Info("fancy-dns started",
		"config", a.cfg.Path(),
		"dns", a.dns.Listeners(),
		"api", a.Addr())
	return nil
}

// startCache opens the bbolt snapshot and restores a warm start.
func (a *App) startCache(cfg *config.Config) error {
	// The memory envelope is a process-wide setting, not a component one, so
	// it is applied on every (re)start and on every SIGHUP reload. Doing it
	// here means a changed limit takes effect with the config that set it.
	applyMemoryLimit(cfg.Memory, a.log)
	opts := cache.Options{
		Size:        cfg.Cache.Size,
		Shards:      cfg.Cache.Shards,
		MaxBytes:    cfg.Memory.EffectiveCacheBytes(),
		MinTTL:      cfg.Cache.MinTTL.D(),
		MaxTTL:      cfg.Cache.MaxTTL.D(),
		NegativeTTL: cfg.Cache.NegativeTTL.D(),
		ServeStale:  cfg.Cache.ServeStale.D(),
		Prefetch:    cfg.Cache.Prefetch,
	}
	if cfg.Cache.Enabled && cfg.Cache.Persist.Enabled {
		st, err := cache.OpenStore(cache.StoreOptions{
			Path:       cfg.CachePath(),
			MaxEntries: cfg.Cache.Persist.MaxEntries,
		})
		if err != nil {
			a.markDegraded("cache persistence: " + err.Error())
			a.log.Warn("cache persistence unavailable, continuing without it", "error", err)
		} else {
			a.cstore = st
			if cfg.Cache.Persist.LoadOnStart {
				lru := cache.NewLRU(cfg.Cache.Size, cfg.Cache.Shards)
				loaded, skipped, lerr := st.Load(lru, time.Now())
				if lerr != nil {
					a.log.Warn("cache snapshot load failed", "error", lerr)
				} else if loaded > 0 {
					a.log.Info("cache snapshot restored", "entries", loaded, "skipped", skipped)
					// Re-seed the real LRU from the temporary one.
					a.cache = cache.NewManager(opts, st)
					lru.Walk(func(k cache.Key, e *cache.Entry) bool {
						a.cache.PutRaw(k, e.Wire, e.TTL, e.Origin, e.Upstream, e.Neg)
						return true
					})
				}
			}
		}
	}
	if a.cache == nil {
		a.cache = cache.NewManager(opts, a.cstore)
	}
	return nil
}

// prewarm resolves a few names at boot in the background, so a restart does
// not look like an outage for the names that matter most.
func (a *App) prewarm(c context.Context, specs []string) {
	go func() {
		for _, spec := range specs {
			name, qtypeStr, ok := strings.Cut(spec, "/")
			if !ok {
				name, qtypeStr = spec, "A"
			}
			qtype, ok := dns.StringToType[strings.ToUpper(qtypeStr)]
			if !ok {
				a.log.Warn("prewarm: unknown record type", "spec", spec)
				continue
			}
			m := new(dns.Msg)
			m.SetQuestion(dns.Fqdn(name), qtype)
			m.RecursionDesired = true
			res := a.engine.Handle(c, engine.Request{Msg: m, Transport: "internal", Start: time.Now()})
			if res.Msg != nil && res.Msg.Rcode == dns.RcodeSuccess {
				a.log.Debug("prewarmed", "name", name, "type", qtypeStr)
			}
		}
	}()
}

// tlsCfgType aliases the TLS config so the listener helpers read cleanly.
type tlsCfgType = tls.Config

// apiTLS builds the admin listener's TLS config when one is configured.
func apiTLS(cfg *config.Config) (*tls.Config, error) {
	if cfg.API.CertFile == "" || cfg.API.KeyFile == "" {
		return nil, nil
	}
	cert, err := tls.LoadX509KeyPair(cfg.API.CertFile, cfg.API.KeyFile)
	if err != nil {
		return nil, fmt.Errorf("load admin certificate: %w", err)
	}
	return &tls.Config{
		Certificates: []tls.Certificate{cert},
		MinVersion:   tls.VersionTLS12,
		NextProtos:   []string{"h2", "http/1.1"},
	}, nil
}

// startLogger starts the access log: JSON lines on stdout for blocked and
// failed queries (and more, if configured). It stores nothing.
func (a *App) startLogger(cfg *config.Config) error {
	if !cfg.AccessLog.Enabled {
		return nil
	}
	a.logger = accesslog.New(accesslog.Options{
		Writer:     a.opt.AccessLogOut,
		Buffer:     cfg.AccessLog.Buffer,
		LogBlocked: cfg.AccessLog.LogBlocked,
		LogErrors:  cfg.AccessLog.LogErrors,
		LogCached:  cfg.AccessLog.LogCached,
		SampleRate: cfg.AccessLog.SampleRate,
	})
	return nil
}

// startBlocklist starts the refresh loop and does the first compile.
func (a *App) startBlocklist() {
	cfg := a.cfg.Get()
	if !cfg.Block.Enabled {
		a.log.Info("blocklists disabled")
		return
	}
	a.blocks = blocklist.NewManager(blocklist.ManagerOptions{
		Config: a.cfg,
		Logger: a.log,
	})
	if a.zones != nil {
		ctx := context.Background()
		content, exists, err := a.zones.Data(ctx, "address-overrides.rpz")
		if err == nil && !exists {
			if seed, readErr := os.ReadFile("/etc/fancy-dns/rpz/address-overrides.rpz"); readErr == nil {
				content = string(seed)
				if saveErr := a.zones.SaveData(ctx, "address-overrides.rpz", content); saveErr != nil {
					a.log.Warn("seed address overrides failed", "error", saveErr)
				}
			}
		}
		if err == nil {
			a.blocks.SetManagedRPZ(content)
		} else {
			a.log.Warn("load address overrides failed", "error", err)
		}
	}
	// Compile once before serving so the first queries are not unprotected.
	if _, err := a.blocks.RefreshAll(context.Background()); err != nil {
		a.markDegraded("blocklist: " + err.Error())
		a.log.Warn("initial blocklist compile failed", "error", err)
	}
}

// startLocalZones compiles the local master files.
func (a *App) startLocalZones(cfg *config.Config) {
	a.local = localzone.NewHolder(nil)
	a.managedZones = localzone.NewManaged(cfg.AbsPath(cfg.Block.ManagedZones))
	a.reloadLocalZones()
}

// managedZoneDir is the directory the console may write zone files into, or
// "" when the console is read-only.
func (a *App) managedZoneDir() string { return a.managedZones.Dir() }

// reloadLocalZones recompiles the local zone store. A failure keeps the
// previous store: serving stale local data beats serving nothing.
func (a *App) reloadLocalZones() {
	a.zoneReloadMu.Lock()
	defer a.zoneReloadMu.Unlock()
	cfg := a.cfg.Get()
	paths := make([]string, 0, len(cfg.Block.LocalZones)+1)
	for _, p := range cfg.Block.LocalZones {
		paths = append(paths, cfg.AbsPath(p))
	}
	// The managed directory is a zone source like any other, so a zone the
	// console created is served by the same store. It is only added when it
	// exists: the loader stats every path it is given, and a directory the
	// console has not written to yet is not a broken configuration.
	if a.managedZones.Exists() {
		paths = append(paths, a.managedZones.Dir())
	}
	if dir := a.kubeZoneDir(); dir != "" {
		paths = append(paths, dir)
	}
	if len(paths) == 0 {
		a.local.Set(nil)
		metrics.SetLocalRecords(0)
		return
	}
	store, err := localzone.Load(paths, localzone.Options{})
	if err != nil {
		a.markDegraded("local zones: " + err.Error())
		a.log.Error("local zone load failed, keeping previous data", "error", err)
		return
	}
	a.local.Set(store)
	metrics.SetLocalRecords(store.Count())
	a.log.Info("local zones loaded", "zones", store.Zones(), "records", store.Count())
}

// startUpstream builds the resolver pool.
func (a *App) startUpstream(cfg *config.Config) {
	a.pool = resolver.NewPoolWithUDPSize(cfg.Upstream, cfg.DNS.MaxUDPSize)
}

// startDNS builds the engine and the listeners.
func (a *App) startDNS(ctx context.Context) {
	a.aclH = acl.NewHolder(acl.Compile(cfgACL(a.cfg.Get())))
	a.reverse = engine.NewReverseCache(engine.ReverseOptions{Max: 8192})
	a.engine = engine.New(engine.Options{
		Config:     a.cfg,
		Logger:     a.logger,
		Blocklists: a.blocksHolder(),
		Cache:      a.cache,
		ACL:        a.aclH,
		Local:      a.local,
		Pool:       a.pool,
		Reverse:    a.reverse,
	})
	a.engine.SetFlusher(a.flush)
	a.dns = server.New(server.Options{
		Engine:  a.engine,
		Config:  a.cfg,
		Logger:  a.log,
		Version: a.ver,
	})
	a.dns.SetIdentity(
		func(ip string) []string { names, _ := a.reverse.Names(ip); return names },
		func(ip string) []string {
			for _, n := range mustNames(a.reverse, ip) {
				if tags := acl.TagsFromName(n); len(tags) > 0 {
					return tags
				}
			}
			return nil
		},
	)
	if err := a.dns.Start(ctx); err != nil {
		a.log.Error("dns listeners failed to start", "error", err)
		a.markDegraded("dns listeners: " + err.Error())
	}
}

func mustNames(rc *engine.ReverseCache, ip string) []string {
	names, _ := rc.Names(ip)
	return names
}

func (a *App) blocksHolder() *blocklist.Holder {
	if a.blocks == nil {
		return blocklist.NewHolder(nil)
	}
	return a.blocks.Holder()
}

func cfgACL(cfg *config.Config) config.ACL { return cfg.ACL }

// startAPI binds the admin listener and, when configured, the embedded UI.
func (a *App) startAPI(ctx context.Context) error {
	cfg := a.cfg.Get()
	if !cfg.API.Enabled {
		return nil
	}
	api.SetMetricsHandler(metrics.Handler())
	hooks := a.hooks()
	srv, err := api.New(hooks)
	if err != nil {
		return err
	}
	a.api = srv

	// One mux: the API claims its base path, the SPA claims the rest. The API
	// is registered first so /api/* never falls through to the SPA's
	// index.html fallback (which would return HTML to a JSON client).
	mux := http.NewServeMux()
	// The api package already registers its routes with the full base path, so
	// no StripPrefix here: a more specific pattern than the SPA's "/" wins, so
	// /api/* never falls through to index.html.
	mux.Handle(cfg.API.BasePath+"/", srv.Handler())
	mux.Handle(cfg.API.BasePath, srv.Handler())
	if cfg.API.UI {
		ui, err := web.Handler("")
		if err != nil {
			a.log.Warn("web console unavailable", "error", err)
		} else {
			mux.Handle("/", ui)
			if web.Available() {
				a.log.Info("web console enabled")
			} else {
				a.log.Warn("web console assets missing from this binary; run 'make web' and rebuild")
			}
		}
	}

	// pprof on the top-level mux would sit OUTSIDE srv.Handler(), so the API's
	// auth would never see it — /debug/pprof/profile starts a CPU profile for
	// anyone who can reach the port. Gate it on loopback instead: profiling
	// means ssh port-forwarding, and that keeps the unauthenticated surface
	// off a routable interface by construction rather than by policy.
	if cfg.API.Pprof {
		if !onlyLoopback(cfg.API.Listen) || !onlyLoopback(cfg.API.TLSListen) {
			a.log.Warn("api.pprof is set but not every admin address is loopback; refusing to mount pprof")
		} else {
			mux.HandleFunc("/debug/pprof/", pprof.Index)
			mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
			mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
			mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
			mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
			a.log.Info("pprof mounted on the admin listener", "path", "/debug/pprof/")
		}
	}

	tlsCfg, tlsErr := apiTLS(cfg)
	var errs []error
	for _, addr := range cfg.API.Listen {
		ln, err := net.Listen("tcp", addr)
		if err != nil {
			errs = append(errs, fmt.Errorf("bind api %s: %w", addr, err))
			continue
		}
		a.startHTTPServer(mux, ln, nil, "api")
	}
	if len(cfg.API.TLSListen) > 0 {
		if tlsErr != nil {
			errs = append(errs, tlsErr)
		} else {
			for _, addr := range cfg.API.TLSListen {
				ln, err := net.Listen("tcp", addr)
				if err != nil {
					errs = append(errs, fmt.Errorf("bind api tls %s: %w", addr, err))
					continue
				}
				a.startHTTPServer(mux, ln, tlsCfg, "api-tls")
			}
		}
	}
	if err := errors.Join(errs...); err != nil {
		a.markDegraded("admin api: " + err.Error())
		a.log.Error("admin api failed to bind", "error", err)
	}
	_ = ctx
	return nil
}

// onlyLoopback reports whether every address is a loopback one. An empty list
// counts as loopback: nothing is bound, so nothing is exposed.
func onlyLoopback(addrs []string) bool {
	for _, a := range addrs {
		host, _, err := net.SplitHostPort(a)
		if err != nil {
			host = a
		}
		ip, err := netip.ParseAddr(host)
		if err != nil {
			// A hostname, or a bare "*": not provably loopback.
			return false
		}
		if !ip.IsLoopback() && !ip.IsUnspecified() {
			return false
		}
	}
	return true
}

func (a *App) startHTTPServer(h http.Handler, ln net.Listener, tlsCfg *tlsCfgType, name string) {
	srv := &http.Server{
		Handler:           h,
		ReadHeaderTimeout: 15 * time.Second,
		IdleTimeout:       120 * time.Second,
	}
	if tlsCfg != nil {
		srv.TLSConfig = tlsCfg
		ln = tls.NewListener(ln, tlsCfg)
	}
	a.apiServers = append(a.apiServers, srv)
	a.apiLn = append(a.apiLn, ln)
	a.log.Info("admin api listening", "addr", ln.Addr().String(), "tls", tlsCfg != nil, "name", name)

	go func() {
		if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) {
			a.log.Error("admin api stopped", "error", err, "name", name)
		}
	}()
}

// startCluster connects to Valkey and wires cluster events into a reload.
func (a *App) startCluster(ctx context.Context, cfg *config.Config) {
	if !cfg.Cluster.Enabled {
		a.cl = cluster.New(cluster.Options{Config: a.cfg, Logger: a.log, Version: a.ver})
		return
	}
	a.cl = cluster.New(cluster.Options{
		Config:    a.cfg,
		Logger:    a.log,
		Version:   a.ver,
		StartedAt: a.startedAt,
		Addr:      cfg.Cluster.Advertise,
		StateFunc: a.nodeState,
		Digests:   a.nodeDigests,
		Handler:   a.onClusterEvent,
	})
	a.cl.Start(ctx)
}

// startBackground runs the periodic jobs: cache snapshots, upstream health
// probes and the metrics gauge.
func (a *App) startBackground(ctx context.Context, cfg *config.Config) {
	go a.watchLocalZones(ctx)
	go a.watchAddressOverrides(ctx)
	// Cache snapshots.
	if a.cstore != nil {
		interval := cfg.Cache.Persist.Interval.D()
		if interval <= 0 {
			interval = 30 * time.Second
		}
		go func() {
			t := time.NewTicker(interval)
			defer t.Stop()
			for {
				select {
				case <-ctx.Done():
					a.snapshotCache()
					return
				case <-t.C:
					a.snapshotCache()
				}
			}
		}()
	}
	// Upstream health probes.
	go a.healthLoop(ctx)
	// The in-flight gauge.
	go func() {
		t := time.NewTicker(2 * time.Second)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				metrics.SetConcurrent(int(a.engine.Inflight()))
			}
		}
	}()
}

func (a *App) snapshotCache() {
	if a.cstore == nil || a.cache == nil {
		return
	}
	n, err := a.cstore.Snapshot(a.cache.LRU(), time.Now())
	if err != nil {
		a.log.Warn("cache snapshot failed", "error", err)
		return
	}
	a.log.Debug("cache snapshot written", "entries", n)
}

func (a *App) healthLoop(ctx context.Context) {
	for {
		interval := a.cfg.Get().Upstream.HealthCheck.D()
		if interval <= 0 {
			interval = 30 * time.Second
		}
		select {
		case <-ctx.Done():
			return
		case <-time.After(interval):
		}
		pool := a.pool
		if pool == nil {
			continue
		}
		hctx, cancel := context.WithTimeout(ctx, 5*time.Second)
		ok := pool.HealthCheck(hctx, 3*time.Second)
		cancel()
		total := len(pool.All())
		for _, u := range pool.All() {
			healthy := u.Health() == resolver.HealthUp
			metrics.SetUpstreamHealth(u.Addr(), u.Config().Proto, healthy)
		}
		if total > 0 && ok == 0 {
			a.markDegraded("no upstream is answering health probes")
		} else {
			a.clearDegraded("no upstream is answering health probes")
		}
	}
}

// markDegraded records a component problem for the health endpoint.
func (a *App) markDegraded(reason string) {
	a.degradedMu.Lock()
	defer a.degradedMu.Unlock()
	for _, r := range a.degraded {
		if r == reason {
			return
		}
	}
	a.degraded = append(a.degraded, reason)
}

func (a *App) clearDegraded(reason string) {
	a.degradedMu.Lock()
	defer a.degradedMu.Unlock()
	out := a.degraded[:0]
	for _, r := range a.degraded {
		if r != reason {
			out = append(out, r)
		}
	}
	a.degraded = out
}

// Degraded returns the current problem list.
func (a *App) Degraded() []string {
	a.degradedMu.RLock()
	defer a.degradedMu.RUnlock()
	return append([]string(nil), a.degraded...)
}

// onConfigSwap reacts to a new configuration.
func (a *App) onConfigSwap(cfg *config.Config) {
	a.log.Info("configuration applied", "version", cfg.Version)
	if a.pool != nil {
		a.pool.Rebind(cfg.Upstream)
		a.invalidateBlocklistCache()
	}
	a.reloadLocalZones()
	if a.blocks != nil && !cfg.Block.Enabled {
		a.log.Info("blocklists disabled by configuration change")
	}
}

// onClusterEvent reacts to a peer message.
func (a *App) onClusterEvent(ctx context.Context, ev cluster.Event) {
	switch ev.Kind {
	case cluster.EventReload:
		a.log.Info("peer requested a config reload", "from", ev.NodeID)
		if err := a.Reload(); err != nil {
			a.log.Warn("cluster-triggered reload failed", "error", err)
		}
	case cluster.EventBlocklists:
		// Only the lease holder fetches, so N instances do not hammer a mirror
		// N times; the others keep serving their compiled copy.
		if a.cl != nil && a.cl.IsLeader() {
			if a.blocks != nil {
				a.blocks.Trigger(firstOr(ev.Sources, ""))
			}
		}
	case cluster.EventConfig:
		// A peer pushed a full document. We do not adopt it silently: a
		// follower adopting a document it cannot see on disk would drift. The
		// operator re-runs a deploy or hits "reload" instead.
		a.log.Info("peer published a new config version", "from", ev.NodeID, "version", ev.Version)
	}
}

func firstOr(s []string, def string) string {
	if len(s) > 0 && s[0] != "" {
		return s[0]
	}
	return def
}

// nodeState supplies the live state published to peers.
func (a *App) nodeState() *cluster.NodeState {
	addrs := make([]string, 0, 4)
	for _, l := range a.dns.Listeners() {
		addrs = append(addrs, l.Transport+":"+l.Addr)
	}
	return &cluster.NodeState{
		Addr:       a.opt.ConfigPath,
		Version:    a.ver,
		StartedAt:  a.startedAt,
		Listeners:  addrs,
		Generation: a.cfg.Generation(),
	}
}

// nodeDigests supplies the digests peers compare.
func (a *App) nodeDigests() (string, int, time.Time, string) {
	var (
		bl    string
		rules int
		built time.Time
	)
	if a.blocks != nil {
		st := a.blocks.State()
		if st != nil {
			bl = st.Digest
			if st.Set != nil {
				rules = st.Set.Len()
				built = st.Set.BuiltAt()
			}
		}
	}
	return bl, rules, built, cluster.Digest(a.cfg.Get())
}

// Reload re-reads the config file and applies it.
func (a *App) Reload() error {
	a.reloadMu.Lock()
	defer a.reloadMu.Unlock()
	return a.cfg.Reload()
}

// Stop shuts everything down in reverse order, with a deadline.
func (a *App) Stop(ctx context.Context) error {
	a.log.Info("shutting down")
	metrics.SetUp(false)
	var errs []error

	if a.cl != nil {
		a.cl.Stop()
	}
	if a.blocks != nil {
		a.blocks.Stop()
	}
	if a.dns != nil {
		if err := a.dns.Stop(ctx); err != nil {
			errs = append(errs, err)
		}
	}
	for _, srv := range a.apiServers {
		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		err := srv.Shutdown(shutCtx)
		cancel()
		if err != nil {
			errs = append(errs, err)
		}
	}
	if err := a.logger.Close(); err != nil {
		errs = append(errs, err)
	}
	if a.cstore != nil {
		// Snapshot, close, then clear the pointer: the background snapshot
		// goroutine also fires on ctx.Done, and a nil check is how it learns
		// to stop rather than writing to a closed file.
		a.snapshotCache()
		if err := a.cstore.Close(); err != nil {
			errs = append(errs, err)
		}
		a.cstore = nil
	}
	if err := errors.Join(errs...); err != nil {
		a.log.Warn("shutdown completed with errors", "error", err)
	} else {
		a.log.Info("shutdown complete")
	}
	return errors.Join(errs...)
}

// Addr returns the admin API address, which a preview needs.
func (a *App) Addr() string {
	for _, ln := range a.apiLn {
		return ln.Addr().String()
	}
	return ""
}

// DataDir is where the bbolt files live.
func (a *App) DataDir() string {
	return filepath.Dir(a.cfg.Get().CachePath())
}

// Log exposes the logger so the command can report signals.
func (a *App) Log() *slog.Logger { return a.log }

// SetAdminAddr overrides the admin API listen address without touching the
// config file, which is what a container platform's -admin-addr flag wants.
func (a *App) SetAdminAddr(addr string) error {
	next := a.cfg.Base()
	cp := *next
	cp.API = next.API
	cp.API.Listen = []string{addr}
	if err := a.cfg.Swap(&cp); err != nil {
		return err
	}
	return nil
}