mesh-catalog/modules/mesh-config/cmd/mesh-config/sources.go

package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"

	"git.novox.be/novox/mesh-sdk/go/delivery"
)

// Reading the sources, writing the resolved values, and publishing the list of predefined keys.

// sources reads everything a resolution reads, from the bus as it is now, from the source the switch names
// (takeover.go): which modules are placed where from the seat mesh-placement, and each placed module's declarations
// from its own manifests as built (issue 1953, ADR 0565 §3); the controller's records are no longer read.
func (c *Config) sources() (Sources, error) {
	s := Sources{Declared: map[string]ModuleDeclarations{}, OnNode: map[string]ModuleDeclarations{},
		Predefined: map[string]Published{}, Values: map[string]Value{}}
	var problems []string
	if err := c.readPlacement(&s); err != nil {
		return s, fmt.Errorf("the seat mesh-placement's placement cannot be read: %w", err)
	}
	if err := c.readDeclarations(&s); err != nil {
		return s, fmt.Errorf("the modules' declarations cannot be read: %w", err)
	}
	for _, u := range s.Unresolved {
		problems = append(problems, fmt.Sprintf("%s on %s: %s", u.Module, u.Node, u.Why))
	}
	if err := c.each(PredefinedState, func(key string, raw []byte) error {
		var p Published
		if err := json.Unmarshal(raw, &p); err != nil {
			return fmt.Errorf("the published predefined key %s is not one: %w", key, err)
		}
		s.Predefined[key] = p
		return nil
	}); err != nil {
		return s, fmt.Errorf("the published list of predefined keys cannot be read: %w", err)
	}
	if err := c.each(ValuesState, func(key string, raw []byte) error {
		var v Value
		if err := json.Unmarshal(raw, &v); err != nil {
			return fmt.Errorf("the value %s is not one: %w", key, err)
		}
		s.Values[key] = v
		return nil
	}); err != nil {
		return s, fmt.Errorf("the values cannot be read: %w", err)
	}
	if err := c.readPulls(&s); err != nil {
		return s, err
	}
	for _, p := range problems {
		c.logf("[mesh-config] a placed module's declarations are not read, so its values are not resolved (they "+
			"stay as last resolved): %s", p)
	}
	return s, nil
}

// readPlacement is every module the seat mesh-placement places on a node (ADR 0567 §1, §9), from the state's keys in
// either form: a node key and its module keys, or the form before, `placement.<node>` (delivery.PlacementsOf;
// research 072 row 37). By node in name order, each node's modules in the order its placement names them.
func (c *Config) readPlacement(s *Sources) error {
	keys := map[string][]byte{}
	if err := c.each(PlacementRead, func(key string, raw []byte) error {
		keys[key] = raw
		return nil
	}); err != nil {
		return err
	}
	placements, err := delivery.PlacementsOf(keys)
	if err != nil {
		return fmt.Errorf("the placement is not read whole: %w", err)
	}
	nodes := make([]string, 0, len(placements))
	for node := range placements {
		nodes = append(nodes, node)
	}
	sort.Strings(nodes)
	for _, node := range nodes {
		for _, module := range placements[node].Assigned() {
			s.Assignments = append(s.Assignments, Assignment{Node: node, Module: module})
		}
	}
	return nil
}

// readDeclarations reads each placed module's declarations from its manifests as the delivery holds them: the trunk's
// (the indexer's `modules.<module>` and its manifest by digest), and, on each node whose want names a build, that
// build's (the builder's `<module>.<fp>` names its manifest digest). On a node the want's build is laid over the
// trunk's, key by key, so a key only the build it runs declares still resolves there and a key the trunk adds resolves
// before its build is wanted; Declared is the trunk's with every key only a wanted build declares. A placed module
// none of whose manifests can be read is Unresolved, with why.
func (c *Config) readDeclarations(s *Sources) error {
	wanted := map[string]string{} // <node>.<module> → the fingerprint its want names
	if err := c.each(WantsRead, func(key string, raw []byte) error {
		node, module, err := delivery.ParseWantKey(key)
		if err != nil {
			return nil // a held key or a node's own want
		}
		var w struct {
			FP      string `json:"fp"`
			Removed bool   `json:"removed"`
		}
		if json.Unmarshal(raw, &w) == nil && !w.Removed && w.FP != "" {
			wanted[node+"."+module] = w.FP
		}
		return nil
	}); err != nil {
		return fmt.Errorf("the assigner's wants: %w", err)
	}
	trunk := map[string]ModuleDeclarations{}
	trunkWhy := map[string]string{}
	for _, module := range modulesOf(s.Assignments) {
		d, err := c.trunkDeclarations(module)
		if err != nil {
			trunkWhy[module] = err.Error()
			continue
		}
		trunk[module] = d
		s.Declared[module] = d
	}
	for _, a := range s.Assignments {
		t, hasTrunk := trunk[a.Module]
		fp, isWanted := wanted[a.Node+"."+a.Module]
		var built ModuleDeclarations
		var builtErr error
		if isWanted {
			built, builtErr = c.builtDeclarations(a.Module, fp)
		}
		switch {
		case isWanted && builtErr == nil:
			on := overlay(t, built)
			s.OnNode[a.Node+"."+a.Module] = on
			d := s.Declared[a.Module]
			s.Declared[a.Module] = overlay(built, d) // the trunk's wins in what a set is judged against
		case hasTrunk:
			if isWanted {
				c.logf("[mesh-config] %s on %s resolves from the trunk's manifest alone: the build its want names "+
					"cannot be read: %v", a.Module, a.Node, builtErr)
			}
		default:
			why := trunkWhy[a.Module]
			if isWanted {
				why += "; and the build its want names: " + builtErr.Error()
			} else {
				why += "; and no want names a build of it on " + a.Node
			}
			s.Unresolved = append(s.Unresolved, Unresolved{Node: a.Node, Module: a.Module, Why: why})
		}
	}
	return nil
}

// overlay is base with every key of over laid on it, the module's name kept.
func overlay(base, over ModuleDeclarations) ModuleDeclarations {
	out := ModuleDeclarations{Module: over.Module, Config: map[string]Declaration{}}
	if out.Module == "" {
		out.Module = base.Module
	}
	for k, d := range base.Config {
		out.Config[k] = d
	}
	for k, d := range over.Config {
		out.Config[k] = d
	}
	return out
}

func modulesOf(as []Assignment) []string {
	seen := map[string]bool{}
	var out []string
	for _, a := range as {
		if !seen[a.Module] {
			seen[a.Module] = true
			out = append(out, a.Module)
		}
	}
	sort.Strings(out)
	return out
}

// trunkDeclarations is a module's declarations at the trunk: the manifest the indexer's `modules.<module>` names.
func (c *Config) trunkDeclarations(module string) (ModuleDeclarations, error) {
	key, err := delivery.ModulesKey(module)
	if err != nil {
		return ModuleDeclarations{}, err
	}
	e, err := c.Store.Get(ModulesRead, key)
	if err != nil {
		return ModuleDeclarations{}, fmt.Errorf("the indexer's modules.%s cannot be read: %w", module, err)
	}
	if e == nil {
		return ModuleDeclarations{}, fmt.Errorf("the indexer holds no modules.%s, so no manifest of it is known", module)
	}
	var m struct {
		Manifest string `json:"manifest"`
	}
	if err := json.Unmarshal(e.Value, &m); err != nil || m.Manifest == "" {
		return ModuleDeclarations{}, fmt.Errorf("the indexer's modules.%s names no manifest", module)
	}
	return c.manifestDeclarations(module, m.Manifest)
}

// builtDeclarations is a module's declarations at a build: the manifest the builder's done key names.
func (c *Config) builtDeclarations(module, fp string) (ModuleDeclarations, error) {
	key, err := delivery.BuilderBuildKey(module, fp)
	if err != nil {
		return ModuleDeclarations{}, err
	}
	e, err := c.Store.Get(BuildsRead, key)
	if err != nil {
		return ModuleDeclarations{}, fmt.Errorf("the builder's %s cannot be read: %w", key, err)
	}
	if e == nil {
		return ModuleDeclarations{}, fmt.Errorf("the builder holds no build %s", key)
	}
	var b struct {
		Manifest string `json:"manifest"`
	}
	if err := json.Unmarshal(e.Value, &b); err != nil || b.Manifest == "" {
		return ModuleDeclarations{}, fmt.Errorf("the builder's %s names no manifest", key)
	}
	return c.manifestDeclarations(module, b.Manifest)
}

// manifestDeclarations is the declarations of the manifest of a digest in the indexer's manifests, read once: a
// manifests key is written once and never changes (ADR 0429 §16), so what it declares is kept by digest.
func (c *Config) manifestDeclarations(module, digest string) (ModuleDeclarations, error) {
	c.manifestsMu.Lock()
	d, ok := c.manifests[digest]
	c.manifestsMu.Unlock()
	if ok {
		return d, nil
	}
	key, err := delivery.ManifestsKey(digest)
	if err != nil {
		return ModuleDeclarations{}, err
	}
	e, err := c.Store.Get(ManifestsRead, key)
	if err != nil {
		return ModuleDeclarations{}, fmt.Errorf("the indexer's manifest %s cannot be read: %w", short(digest), err)
	}
	if e == nil {
		return ModuleDeclarations{}, fmt.Errorf("the indexer holds no manifest %s", short(digest))
	}
	var m struct {
		Module string `json:"module"`
		JSON   string `json:"manifest"`
	}
	if err := json.Unmarshal(e.Value, &m); err != nil || m.JSON == "" {
		return ModuleDeclarations{}, fmt.Errorf("the indexer's manifest %s is not one", short(digest))
	}
	if m.Module != module {
		return ModuleDeclarations{}, fmt.Errorf("the indexer's manifest %s is %s's, not %s's", short(digest), m.Module, module)
	}
	d, err = ManifestDeclarations([]byte(m.JSON))
	if err != nil {
		return ModuleDeclarations{}, fmt.Errorf("the manifest %s: %w", short(digest), err)
	}
	c.manifestsMu.Lock()
	if c.manifests == nil {
		c.manifests = map[string]ModuleDeclarations{}
	}
	c.manifests[digest] = d
	c.manifestsMu.Unlock()
	return d, nil
}

// each hands every key of a state to visit.
func (c *Config) each(state string, visit func(key string, raw []byte) error) error {
	keys, err := c.Store.Keys(state)
	if err != nil {
		return err
	}
	for _, k := range keys {
		e, err := c.Store.Get(state, k)
		if err != nil {
			return err
		}
		if e == nil {
			continue // removed between the listing and the read
		}
		if err := visit(k, e.Value); err != nil {
			return err
		}
	}
	return nil
}

// held is what mesh-config_resolved holds now.
func (c *Config) held() (map[string]Resolved, error) {
	out := map[string]Resolved{}
	err := c.each(ResolvedState, func(key string, raw []byte) error {
		var r Resolved
		if err := json.Unmarshal(raw, &r); err != nil {
			// A value nobody can read is written again from the sources.
			out[key] = Resolved{Key: key, Source: "unreadable"}
			return nil
		}
		out[key] = r
		return nil
	})
	return out, err
}

// writeResolved makes mesh-config_resolved hold what the sources resolve to, writing only what moved, and
// answers what moved.
func (c *Config) writeResolved(s Sources) ([]Moved, error) {
	held, err := c.held()
	if err != nil {
		return nil, fmt.Errorf("the resolved values cannot be read: %w", err)
	}
	want := Resolve(s)
	// A placed module whose declarations cannot be read keeps what was resolved for it: not knowing is no reason to take
	// a node's values away (issue 1953).
	for key, h := range held {
		if _, ok := want[key]; !ok && h.Source != "unreadable" && s.unresolved(h.Node, h.Module) {
			want[key] = h
		}
	}
	moved := diff(held, want)
	var problems []error
	for _, m := range moved {
		key := ResolvedKey(m.Node, m.Module, m.Key)
		if m.Gone {
			if err := c.Store.Delete(ResolvedState, key); err != nil {
				problems = append(problems, fmt.Errorf("removing %s: %w", key, err))
			}
			continue
		}
		if err := c.Store.Put(ResolvedState, key, want[key]); err != nil {
			problems = append(problems, fmt.Errorf("writing %s: %w", key, err))
		}
	}
	// A key that is no resolved value at all (written by an older mesh-config, or unreadable) is removed.
	for key, h := range held {
		if _, ok := want[key]; !ok && h.Source == "unreadable" {
			if err := c.Store.Delete(ResolvedState, key); err != nil {
				problems = append(problems, fmt.Errorf("removing %s: %w", key, err))
			}
		}
	}
	return moved, errors.Join(problems...)
}

// Resolution is what one pass of Reconcile did.
type Resolution struct {
	Moved     []Moved `json:"moved"`
	Published string  `json:"published,omitempty"`
	Withheld  string  `json:"withheld,omitempty"`
	// Declared are the modules whose `declared` key was written, and Candidates what the indexer's config candidates
	// did after the take-over.
	Declared   []string `json:"declared,omitempty"`
	Candidates []string `json:"candidates,omitempty"`
	// SourceGone is said while the stored switch still names the controller, whose records are no longer read.
	SourceGone string `json:"source_gone,omitempty"`
}

// Reconcile publishes the newest list of predefined keys and brings mesh-config_resolved to what the sources
// resolve to: run at start, and whenever an assignment, a declaration or the list moves. A move it makes raises
// config-changed like a set, with no level, naming what moved.
func (c *Config) Reconcile() (Resolution, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	var r Resolution
	if sw, serr := c.source(); serr == nil && sw.Gone != "" {
		r.SourceGone = sw.Gone
		c.logf("[mesh-config] %s", sw.Gone)
	}
	published, err := c.publish()
	if err != nil {
		c.logf("[mesh-config] the list of predefined keys is not published: %v", err)
	}
	r.Published, r.Withheld = published, c.withheld
	s, err := c.sources()
	if err != nil {
		return r, err
	}
	moved, err := c.writeResolved(s)
	r.Moved = moved
	if len(moved) > 0 {
		c.record("resolved", Change{Action: "resolve", Why: "an assignment, a declaration or the list of predefined " +
			"keys changed", At: c.now(), Moved: moved})
	}
	if written, derr := c.writeDeclared(s); derr != nil {
		c.logf("[mesh-config] declared is not all written: %v", derr)
	} else if len(written) > 0 {
		r.Declared = written
	}
	if sw, serr := c.source(); serr == nil && sw.Source == SourceIndexer {
		r.Candidates = c.takeCandidates()
	}
	return r, err
}

// publish makes mesh-config_predefined hold the indexer's list (the controller's mirror is no longer read, issue 774)
// when it is newer than the one published (ADR 0348 §7: the newest trunk build wins;
// the indexer never keeps an older merge's, so its list is newer when its commit differs). **A list dropping a key that still holds a value is withheld**: the
// list published stays, and why is said in every listing and in the log, because a value for a key no longer on
// the list could neither be resolved nor cleared by name (§7: removing a key still holding a value is refused).
// It answers what it published, empty when nothing changed.
func (c *Config) publish() (string, error) {
	list, found, err := c.indexerList()
	if err != nil {
		return "", fmt.Errorf("the indexer's list cannot be read: %w", err)
	}
	if !found {
		return "", nil // no list kept yet
	}
	current := map[string]Published{}
	if err := c.each(PredefinedState, func(key string, raw []byte) error {
		var p Published
		if err := json.Unmarshal(raw, &p); err != nil {
			return err
		}
		current[key] = p
		return nil
	}); err != nil {
		return "", err
	}
	generation, sameCommit := 0, len(current) > 0
	for _, p := range current {
		generation = max(generation, p.Generation)
		sameCommit = sameCommit && p.Commit == list.Commit
	}
	if len(current) > 0 && (sameCommit || (list.Generation > 0 && generation >= list.Generation)) {
		c.withheld = ""
		return "", nil
	}
	held := map[string][]string{}
	if err := c.each(ValuesState, func(key string, _ []byte) error {
		level, k, ok := parseValueKey(key)
		if ok && level.Kind != LevelModule {
			if _, kept := list.Keys[k]; !kept {
				held[k] = append(held[k], level.String())
			}
		}
		return nil
	}); err != nil {
		return "", err
	}
	if len(held) > 0 {
		var why []string
		for _, k := range sortedKeys(held) {
			why = append(why, fmt.Sprintf("%s (still set as %s)", k, strings.Join(held[k], ", ")))
		}
		c.withheld = fmt.Sprintf("the list at %s (generation %d) is not published: it drops %s; clear those values "+
			"first, or keep the key (novox/hq ADR 0348 §7)", short(list.Commit), list.Generation, strings.Join(why, "; "))
		c.logf("[mesh-config] %s", c.withheld)
		return "", nil
	}
	c.withheld = ""
	var problems []error
	for _, k := range sortedKeys(list.Keys) {
		p := Published{PredefinedKey: list.Keys[k], Generation: list.Generation, Commit: list.Commit}
		if err := c.Store.Put(PredefinedState, k, p); err != nil {
			problems = append(problems, err)
		}
	}
	for k := range current {
		if _, kept := list.Keys[k]; !kept {
			if err := c.Store.Delete(PredefinedState, k); err != nil {
				problems = append(problems, err)
			}
		}
	}
	if err := errors.Join(problems...); err != nil {
		return "", err
	}
	return fmt.Sprintf("the list at %s (generation %d): %s", short(list.Commit), list.Generation,
		joinKeys(list.Keys)), nil
}

func short(commit string) string {
	if len(commit) > 8 {
		return commit[:8]
	}
	return commit
}