fancy-dns/internal/app/hooks.go

package app

import (
	"context"
	"errors"
	"fmt"
	"net/netip"
	"reflect"
	"sort"
	"strings"
	"time"

	"github.com/miekg/dns"

	"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/config"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/engine"
	"git.k8s-git-k6n.net/mats/fancy-dns/internal/resolver"
)

// hooks builds the api.Hooks that connect the admin API to the live
// components. Everything the UI can read or change goes through here.
func (a *App) hooks() api.Hooks {
	return api.Hooks{
		Store:             a.cfg,
		Logger:            a.log,
		Version:           a.ver,
		StartedAt:         a.startedAt,
		Status:            a.status,
		Health:            a.health,
		BlocklistState:    a.blocklistState,
		BlocklistStats:    a.blocklistStats,
		RefreshBlocklists: a.refreshBlocklists,
		ToggleBlocklist:   a.toggleBlocklist,
		DomainCheck:       a.domainCheck,
		Resolve:           a.resolveTest,
		ResolveUpstream:   a.resolveTestUpstream,
		ListDomains:       a.listDomains,
		AddDomain:         a.addDomain,
		DeleteDomain:      a.deleteDomain,
		ListZones:         a.listZones,
		GetACL:            a.getACL,
		PutACL:            a.putACL,
		GetUpstream:       a.getUpstreams,
		PutUpstream:       a.putUpstream,
		CacheStats:        a.cacheStats,
		FlushCache:        a.flushCache,
		Flush:             a.flush,
	}
}

func (a *App) resolveTestUpstream(ctx context.Context, name, qtype, upstream string) (any, error) {
	if a.pool == nil {
		return nil, errors.New("upstream pool is not running")
	}
	up := a.pool.Get(strings.TrimSpace(upstream))
	if up == nil {
		return nil, fmt.Errorf("unknown upstream %q", upstream)
	}
	qt := dns.TypeA
	if qtype != "" {
		var ok bool
		qt, ok = dns.StringToType[strings.ToUpper(strings.TrimSpace(qtype))]
		if !ok {
			return nil, fmt.Errorf("unknown record type %q", qtype)
		}
	}
	q := new(dns.Msg)
	q.SetQuestion(dns.Fqdn(name), qt)
	q.RecursionDesired = true
	started := time.Now()
	resp, addr, err := a.pool.Exchange(ctx, q, []string{up.Addr()}, 5*time.Second)
	if err != nil {
		return nil, err
	}
	out := map[string]any{"name": dns.Fqdn(name), "type": dns.TypeToString[qt], "rcode": "SERVFAIL", "action": "upstream", "duration_ms": time.Since(started).Milliseconds(), "upstream": addr, "group": "", "cache": "bypass", "local": false, "truncated": false, "blocked": false, "block_source": "", "answers": []string{}}
	if resp != nil {
		out["rcode"] = dns.RcodeToString[resp.Rcode]
		ans := []string{}
		for _, rr := range resp.Answer {
			ans = append(ans, strings.TrimSpace(rr.String()))
		}
		out["answers"] = ans
	}
	return out, nil
}

// health is the unauthenticated probe payload.
func (a *App) health() any {
	status := "ok"
	problems := a.Degraded()
	switch {
	case len(problems) > 0:
		status = "degraded"
	}
	if a.dns == nil || len(a.dns.Listeners()) == 0 {
		status = "down"
	}
	if a.logger.Dropped() > 0 && status == "ok" {
		// Lines are dropped when the log stream cannot keep up. DNS is fine, but
		// whoever reads the log is missing events, which is worth a mention.
		status = "degraded"
		problems = append(problems, "access log is dropping lines")
	}
	return map[string]any{
		"status":         status,
		"version":        a.ver,
		"uptime_s":       int64(time.Since(a.startedAt).Seconds()),
		"config_version": a.cfg.Get().Version,
		"problems":       problems,
	}
}

// status is the per-component runtime view.
func (a *App) status() any {
	out := map[string]any{
		"listeners":  a.dns.Listeners(),
		"transports": a.dns.ByTransport(),
		"degraded":   a.Degraded(),
	}
	if a.blocks != nil {
		st := a.blocks.State()
		if st != nil {
			out["blocklist"] = map[string]any{
				"rules":      st.Set.Len(),
				"sources":    len(st.Sources),
				"duration":   st.Duration,
				"built_at":   st.Set.BuiltAt(),
				"errors":     st.Errors,
				"digest":     st.Digest,
				"has_wild":   st.Set.HasWildcards(),
				"duplicates": st.Set.DuplicatesN(),
			}
		}
	}
	if a.local != nil {
		if store := a.local.Get(); store != nil {
			out["local_zones"] = map[string]any{
				"zones":   store.Zones(),
				"records": store.Count(),
				"files":   store.Files(),
			}
		}
	}
	if a.cache != nil {
		out["cache"] = a.cacheStats()
	}
	out["access_log"] = map[string]any{
		"enabled": a.logger != nil,
		"dropped": a.logger.Dropped(),
	}
	if a.pool != nil {
		out["upstream"] = a.getUpstreams()
	}
	if a.aclH != nil {
		out["acl"] = a.getACL()
	}
	if a.cl != nil {
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		out["cluster"] = a.cl.Status(ctx)
		cancel()
	}
	if a.engine != nil {
		out["engine"] = map[string]any{
			"inflight":    a.engine.Inflight(),
			"dropped":     a.engine.Dropped(),
			"group_stats": a.engine.GroupStats(),
		}
	}
	return out
}

// blocklistState is the full per-source view for the UI.
func (a *App) blocklistState() any {
	if a.blocks == nil {
		return map[string]any{
			"items": []any{},
			"stats": map[string]any{"total": 0},
		}
	}
	st := a.blocks.State()
	items := make([]blocklist.Source, 0, len(st.Sources))
	for _, s := range st.Sources {
		items = append(items, s)
	}
	sort.Slice(items, func(i, j int) bool { return items[i].Name < items[j].Name })
	return map[string]any{
		"items":  items,
		"stats":  blocklistSourceStats(st),
		"errors": st.Errors,
		"digest": st.Digest,
	}
}

func blocklistSourceStats(st *blocklist.State) map[string]any {
	if st == nil || st.Set == nil {
		return map[string]any{"total": 0}
	}
	s := st.Set.Stats()
	return map[string]any{
		"total":           s.Total,
		"exact":           s.Exact,
		"subdomain_rules": s.Subdomains,
		"root_rules":      s.RootRules,
		"allow":           s.Allow,
		"block":           s.Block,
		"passthrough":     s.Passthru,
		"drop":            s.Drop,
		"duplicates":      s.Duplicates,
		"sources":         s.Sources,
		"built_at":        s.BuiltAt,
		"sources_detail":  s.PerSource,
	}
}

func (a *App) blocklistStats() any {
	if a.blocks == nil {
		return map[string]any{"total": 0}
	}
	return blocklistSourceStats(a.blocks.State())
}

// invalidateBlocklistCache drops every cached answer whose verdict the new
// rules might change. Without this, a name that has just become blocked keeps
// serving its cached good answer until the TTL expires, which is exactly the
// bug an operator hits right after clicking "add rule".
func (a *App) invalidateBlocklistCache() {
	if a.cache == nil {
		return
	}
	a.cache.Flush()
}

// refreshBlocklists recompiles the list sources.
func (a *App) refreshBlocklists(ctx context.Context, names []string) error {
	if a.blocks == nil {
		return fmt.Errorf("blocklists are disabled")
	}
	a.invalidateBlocklistCache()
	go func() {
		if len(names) == 0 {
			if _, err := a.blocks.ForceRefresh(ctx); err != nil {
				a.log.Warn("blocklist refresh failed", "error", err)
			}
			return
		}
		// The manager compiles everything in one pass, which is cheaper than
		// a pass per name and keeps the compiled set internally consistent.
		if _, err := a.blocks.ForceRefresh(ctx); err != nil {
			a.log.Warn("blocklist refresh failed", "error", err, "names", names)
		}
	}()
	// Tell the cluster leader so every instance ends up on the same content.
	if a.cl != nil {
		_ = a.cl.BroadcastRefresh(ctx, names)
	}
	return nil
}

// toggleBlocklist flips one source's enabled flag and persists it.
func (a *App) toggleBlocklist(name string, enabled bool) error {
	if a.blocks == nil {
		return fmt.Errorf("blocklists are disabled")
	}
	if a.clusterOwned("list", name) {
		return fmt.Errorf("blocklist %q is a Kubernetes BlockList; set spec.disabled on the object instead", name)
	}
	err := a.mutateConfig(func(c *config.Config) error {
		found := false
		for i := range c.Block.Lists {
			if c.Block.Lists[i].Name == name {
				c.Block.Lists[i].Enabled = enabled
				found = true
			}
		}
		if !found {
			return fmt.Errorf("no blocklist named %q", name)
		}
		return nil
	})
	if err != nil {
		return err
	}
	a.invalidateBlocklistCache()
	go func() {
		if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil {
			a.log.Warn("refresh after toggle failed", "error", rerr)
		}
	}()
	return nil
}

// mutateConfig clones, mutates, validates and commits the config, returning
// the validation error when the change is rejected.
func (a *App) mutateConfig(fn func(*config.Config) error) error {
	// Start from the base, not the effective config: the effective one holds
	// the cluster's groups and lists, and persisting those would copy them
	// into the YAML file where they would shadow their own objects.
	cur := a.cfg.Base()
	next := *cur
	// A shallow copy is not enough: the slices and maps we touch need their own
	// backing arrays, or a rejected change would corrupt the live config.
	next.Block = cur.Block
	next.Block.Lists = append([]config.ListSource(nil), cur.Block.Lists...)
	next.Block.LocalZones = append([]string(nil), cur.Block.LocalZones...)
	next.Block.BlockEntries = append([]string(nil), cur.Block.BlockEntries...)
	next.Block.AllowEntries = append([]string(nil), cur.Block.AllowEntries...)
	next.ACL = cur.ACL
	next.ACL.Groups = append([]config.Group(nil), cur.ACL.Groups...)
	next.Upstream = cur.Upstream
	next.Upstream.Servers = append([]config.UpstreamServer(nil), cur.Upstream.Servers...)
	next.Upstream.Bootstraps = append([]config.UpstreamBootstrap(nil), cur.Upstream.Bootstraps...)

	if err := fn(&next); err != nil {
		return err
	}
	if err := next.Validate(); err != nil {
		return err
	}
	if a.api == nil {
		return a.cfg.Swap(&next)
	}
	return a.api.Commit(&next)
}

// domainCheck explains how a name would be treated.
func (a *App) domainCheck(name string) (any, error) {
	entry := (*blocklist.Entry)(nil)
	if a.blocks != nil {
		if set := a.blocks.Holder().Get(); set != nil {
			entry = set.Match(name)
		}
	}
	out := map[string]any{
		"name":  name,
		"query": strings.TrimSuffix(name, "."),
	}
	if entry == nil {
		out["matched"] = false
		out["action"] = "none"
		out["source"] = ""
		return out, nil
	}
	kind := "exact"
	pattern := entry.Name
	if set := a.blocks.Holder().Get(); set != nil {
		// Recover which of the three maps matched so the UI can say *why*.
		switch set.MatchPattern(name) {
		case blocklist.PatternSub:
			kind = "subdomain"
			pattern = "*." + entry.Name
		case blocklist.PatternRoot:
			kind = "domain"
			pattern = "**." + entry.Name
		}
	}
	out["matched"] = true
	out["action"] = entry.Action.String()
	out["source"] = entry.Source
	out["categories"] = entry.Categories
	out["hits"] = entry.Hits()
	out["rule"] = pattern
	out["matched_pattern"] = pattern
	out["match_kind"] = kind
	return out, nil
}

// resolveTest runs one query through the live pipeline. It goes through the
// engine rather than straight to an upstream: a test that skipped the
// pipeline would report a healthy upstream while the blocklist, the ACL or the
// local zones were what a client actually hit.
func (a *App) resolveTest(ctx context.Context, name, qtype string) (any, error) {
	if a.engine == nil {
		return nil, errors.New("the query engine is not running")
	}
	qt := dns.TypeA
	if qtype = strings.TrimSpace(qtype); qtype != "" {
		var ok bool
		qt, ok = dns.StringToType[strings.ToUpper(qtype)]
		if !ok {
			return nil, fmt.Errorf("unknown record type %q", qtype)
		}
	}
	q := new(dns.Msg)
	q.SetQuestion(dns.Fqdn(name), qt)
	q.RecursionDesired = true
	res := a.engine.Handle(ctx, engine.Request{
		Msg:       q,
		Remote:    netip.AddrFrom4([4]byte{127, 0, 0, 1}),
		Transport: "api",
		Start:     time.Now(),
	})
	out := map[string]any{
		"name":         dns.Fqdn(name),
		"type":         dns.TypeToString[qt],
		"rcode":        "SERVFAIL",
		"action":       string(res.Action),
		"duration_ms":  res.Duration.Milliseconds(),
		"upstream":     res.Upstream,
		"group":        res.Group,
		"cache":        res.CacheKind,
		"local":        res.Local,
		"truncated":    res.Truncated,
		"blocked":      res.BlockSource != "",
		"block_source": res.BlockSource,
		"answers":      []string{},
	}
	if res.Msg == nil {
		out["error"] = "no response"
		return out, nil
	}
	out["rcode"] = dns.RcodeToString[res.Msg.Rcode]
	answers := make([]string, 0, len(res.Msg.Answer))
	for _, rr := range res.Msg.Answer {
		answers = append(answers, strings.TrimSpace(rr.String()))
	}
	out["answers"] = answers
	return out, nil
}

// listDomains pages the compiled rules.
// request would be too slow, so the filter runs over a bounded sample unless
// the query is a prefix/exact lookup, which the LRU can answer directly.
func (a *App) listDomains(q api.DomainQuery) (any, error) {
	if a.blocks == nil {
		return map[string]any{"items": []any{}, "total": 0, "page": 1, "per_page": q.PerPage}, nil
	}
	set := a.blocks.Holder().Get()
	if set == nil {
		return map[string]any{"items": []any{}, "total": 0, "page": 1, "per_page": q.PerPage}, nil
	}
	if q.Text != "" && (q.Action == "" && q.List == "") {
		if e := set.Lookup(q.Text); e != nil {
			return map[string]any{
				"items":    []blocklist.Entry{*e},
				"total":    1,
				"page":     1,
				"per_page": q.PerPage,
			}, nil
		}
	}
	type row struct {
		Name       string   `json:"name"`
		Action     string   `json:"action"`
		Source     string   `json:"source"`
		Categories []string `json:"categories"`
		Hits       uint64   `json:"hits"`
		Rule       string   `json:"rule"`
	}
	text := strings.ToLower(strings.TrimSpace(q.Text))
	action := strings.ToLower(q.Action)
	// The cap keeps a request bounded on a large set; the UI pages anyway.
	const scanCap = 400000
	matches := make([]row, 0, 256)
	total := 0
	set.Each(func(pattern string, e *blocklist.Entry, kind blocklist.PatternKind) bool {
		if len(matches) >= scanCap {
			return false
		}
		if action != "" && e.Action.String() != action {
			return true
		}
		if q.List != "" && e.Source != q.List {
			return true
		}
		if text != "" && !strings.Contains(pattern, text) {
			return true
		}
		total++
		matches = append(matches, row{
			Name:       pattern,
			Action:     e.Action.String(),
			Source:     e.Source,
			Categories: e.Categories,
			Hits:       e.Hits(),
			// The wire form, so the console can re-add this exact rule
			// (with its wildcard) instead of degrading it to an exact name.
			Rule: kind.WireName(pattern),
		})
		return true
	})
	sort.Slice(matches, func(i, j int) bool {
		if matches[i].Hits != matches[j].Hits {
			return matches[i].Hits > matches[j].Hits
		}
		return matches[i].Name < matches[j].Name
	})
	start := (q.Page - 1) * q.PerPage
	if start > len(matches) {
		start = len(matches)
	}
	end := start + q.PerPage
	if end > len(matches) {
		end = len(matches)
	}
	return map[string]any{
		"items":     matches[start:end],
		"total":     total,
		"page":      q.Page,
		"per_page":  q.PerPage,
		"truncated": total > len(matches),
	}, nil
}

// addDomain appends an inline rule and persists it.
func (a *App) addDomain(_ context.Context, in api.DomainInput) (any, error) {
	name := strings.TrimSpace(in.Name)
	if name == "" {
		return nil, fmt.Errorf("name is required")
	}
	act, ok := blocklist.ParseAction(in.Action)
	if !ok {
		return nil, fmt.Errorf("unknown action %q", in.Action)
	}
	// A bulk paste goes in as several entries; a single name goes in as one.
	entries := []string{}
	if in.Content != "" {
		entries = strings.Split(in.Content, "\n")
	} else {
		entries = []string{name}
	}
	err := a.mutateConfig(func(c *config.Config) error {
		for _, e := range entries {
			e = strings.TrimSpace(e)
			if e == "" || strings.HasPrefix(e, "#") {
				continue
			}
			entry := e
			if act == blocklist.ActionBlock {
				c.Block.BlockEntries = append(c.Block.BlockEntries, entry)
			} else {
				c.Block.AllowEntries = append(c.Block.AllowEntries, entry)
			}
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	if a.blocks != nil {
		if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil {
			a.log.Warn("refresh after add failed", "error", rerr)
		}
	}
	// A new rule changes the verdict for the whole subtree it covers, and an
	// allow rule can rescue a name many other lists blocked, so the honest
	// thing is to drop the affected answers rather than guess.
	a.invalidateBlocklistCache()
	return map[string]any{"ok": true, "name": name, "count": len(entries)}, nil
}

// deleteDomain removes a rule and persists the change.
func (a *App) deleteDomain(_ context.Context, name, action, list string) error {
	target := strings.ToLower(strings.TrimSpace(name))
	removed := 0
	err := a.mutateConfig(func(c *config.Config) error {
		strip := func(entries []string) []string {
			out := entries[:0]
			for _, e := range entries {
				normal := strings.ToLower(strings.TrimSpace(e))
				normal = strings.TrimPrefix(strings.TrimPrefix(normal, "*."), "**.")
				normal = strings.TrimSuffix(normal, ".")
				if normal == target {
					removed++
					continue
				}
				out = append(out, e)
			}
			return out
		}
		c.Block.BlockEntries = strip(c.Block.BlockEntries)
		c.Block.AllowEntries = strip(c.Block.AllowEntries)
		// A file-backed list is edited by editing the file, so the API only
		// reports that rather than pretending it removed something.
		if list != "" && removed == 0 {
			return fmt.Errorf("%q comes from list %q; edit that list instead", name, list)
		}
		if removed == 0 {
			return fmt.Errorf("no inline rule matches %q", name)
		}
		return nil
	})
	if err != nil {
		return err
	}
	if a.blocks != nil {
		if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil {
			a.log.Warn("refresh after delete failed", "error", rerr)
		}
	}
	a.invalidateBlocklistCache()
	return nil
}

// getACL returns the ACL groups.
func (a *App) getACL() any {
	groups := a.engine.GroupStats()
	if a.kubeSnap.Load() != nil {
		inFile := map[string]bool{}
		for _, g := range a.cfg.Base().ACL.Groups {
			inFile[g.Name] = true
		}
		for i := range groups {
			if !inFile[groups[i].Name] && a.clusterOwned("group", groups[i].Name) {
				groups[i].Source = "kubernetes"
			}
		}
	}
	return map[string]any{
		"groups":             groups,
		"default_policy":     a.cfg.Get().ACL.DefaultPolicy,
		"default_rate_limit": a.cfg.Get().ACL.DefaultRateLimit,
	}
}

// putACL validates and persists new ACL groups.
func (a *App) putACL(_ context.Context, groups []config.Group, def string) (int64, error) {
	// The console shows the effective list, so a save sends the cluster's
	// groups back along with the file's. Unchanged ones are dropped here (the
	// cluster still owns them); a changed one is refused, because the write
	// would land in the YAML and silently shadow the object.
	if snap := a.kubeSnap.Load(); snap != nil {
		owned := make(map[string]config.Group, len(snap.Groups))
		for _, g := range snap.Groups {
			owned[g.Name] = g
		}
		kept := groups[:0:0]
		for _, g := range groups {
			cg, ok := owned[g.Name]
			if !ok {
				kept = append(kept, g)
				continue
			}
			if !reflect.DeepEqual(g, cg) {
				return 0, fmt.Errorf("group %q is a Kubernetes ClientGroup; edit the object instead", g.Name)
			}
		}
		groups = kept
	}
	err := a.mutateConfig(func(c *config.Config) error {
		c.ACL.Groups = groups
		if def != "" {
			c.ACL.DefaultPolicy = def
		}
		return nil
	})
	if err != nil {
		return 0, err
	}
	// An allow_domains or deny_domains change alters the verdict for names
	// already in the cache, so the cached answers are no longer trustworthy.
	if a.cache != nil {
		a.cache.Flush()
	}
	return a.cfg.Get().Version, nil
}

// getUpstreams returns the pool state.
func (a *App) getUpstreams() any {
	cfg := a.cfg.Get()
	// Keep forwarding targets out of the editable default-server list.
	defaults := a.pool.Defaults()
	servers := make([]resolver.Stats, 0, len(defaults))
	for _, u := range defaults {
		servers = append(servers, u.Stats())
	}
	return map[string]any{
		"policy":  cfg.Upstream.Policy,
		"servers": servers,
		// Durations on the wire are duration strings, everywhere: this is the
		// same spelling config.yaml uses and the same one Duration.MarshalJSON
		// writes for bootstraps[].refresh. It used to be a float of seconds
		// while the console declared the field a string.
		"bootstraps":           cfg.Upstream.Bootstraps,
		"healthcheck_interval": cfg.Upstream.HealthCheck,
		"max_attempts":         cfg.Upstream.MaxAttempts,
		"breaker_threshold":    cfg.Upstream.BreakerThreshold,
		"breaker_cooldown":     cfg.Upstream.BreakerCooldown,
	}
}

// putUpstream validates and persists a new upstream config.
func (a *App) putUpstream(_ context.Context, up config.Upstream) (int64, error) {
	err := a.mutateConfig(func(c *config.Config) error {
		// The console edits default servers; preserve YAML-only forwarding rules.
		up.ForwardZones = c.Upstream.ForwardZones
		c.Upstream = up
		return nil
	})
	if err != nil {
		return 0, err
	}
	// The pool rebinds itself on the config swap; do it now so the UI sees
	// the new servers immediately rather than after the next reload.
	a.pool.Rebind(up)
	return a.cfg.Get().Version, nil
}

// cacheStats returns the cache snapshot.
func (a *App) cacheStats() any {
	if a.cache == nil {
		return map[string]any{"enabled": false}
	}
	st := a.cache.Stats()
	return map[string]any{
		"enabled":         st.Enabled,
		"size":            st.Size,
		"entries":         st.Entries,
		"shards":          st.Shards,
		"min_ttl_ms":      st.MinTTLMS,
		"max_ttl_ms":      st.MaxTTLMS,
		"serve_stale_ms":  st.ServeStaleM,
		"negative_ttl_ms": st.NegativeMS,
		"hit_rate":        st.HitRate,
		"hits":            st.Hits,
		"misses":          st.Misses,
		"stale_hits":      st.StaleHits,
		"evictions":       st.Evictions,
		"bytes":           st.Bytes,
		"bytes_human":     st.MemoryHuman,
		"persisted":       st.Persist,
	}
}

// flushCache empties the cache.
func (a *App) flushCache() int64 {
	if a.cache == nil {
		return 0
	}
	st := a.cache.Stats()
	a.cache.Flush()
	return int64(st.Entries)
}

// flush persists buffered state.
func (a *App) flush() {
	a.logger.Flush()
	a.snapshotCache()
}