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

// mesh-config: the owner of every config value, holder of the mesh-config seat (novox/hq ADR 0348 §9, design 57
// K3). A Go bundle served by a runtime of its own, on its own bus account (`runs-as`, ADR 0259 §8), so that its
// account alone writes its buckets: mesh-config_values (each value set, at the mesh, a node, or a module on a
// node), mesh-config_resolved (each module's value on each node, with its source), mesh-config_predefined (the
// published list of predefined keys) and mesh-config_proposals (each value proposed for the operator's Approve,
// asked under this module's own name, design 57 K4).
// It resolves from where the seat mesh-placement places each module, and each module's declarations as its own
// manifests declare them, the trunk's and its wanted build's (the indexer's and the builder's state, issue 1953); the
// list of predefined keys comes from the indexer's (ADR 0565 §2); the controller's records are no longer read (issue 774). stdout is the MCP channel; what this module says, it says on stderr.
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"strings"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/handacts"
)

// Seat is the role this module holds, once for the mesh, and Verb its one verb.
const (
	Seat = "mesh-config"
	Verb = "config"
)

func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }

// theViews are the pictures of the delivery owners' state, made when the module runs (nil in a test of the tools).
var theViews *ownerViews

// runtimeStore is the bus, through the runtime that launched this bundle. A state of the delivery owners that the
// owners' tools answer is read from the picture those keep (ownerviews.go), the rest from the bus.
type runtimeStore struct{}

func (runtimeStore) Get(state, key string) (*Entry, error) {
	if theViews != nil {
		if e, followed, err := theViews.get(state, key); followed {
			return e, err
		}
	}
	e, err := stdio.State(state).Get(key)
	if err != nil || e == nil {
		return nil, err
	}
	return &Entry{Key: e.Key, Value: e.Value, Revision: e.Revision}, nil
}

func (runtimeStore) Put(state, key string, value any) error {
	_, err := stdio.State(state).Put(key, value)
	return err
}

func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) }

func (runtimeStore) Keys(state string) ([]string, error) {
	if theViews != nil {
		if keys, followed, err := theViews.keys(state); followed {
			return keys, err
		}
	}
	return stdio.State(state).Keys()
}

func (runtimeStore) Append(log, key string, value any) error {
	_, err := stdio.Log(log).Append(key, value)
	return err
}

func (runtimeStore) Read(log string) ([]LogEntry, error) {
	var out []LogEntry
	for from := uint64(0); ; {
		page, err := stdio.Log(log).Read(from, "", 1000)
		if err != nil {
			return nil, err
		}
		for _, e := range page.Entries {
			out = append(out, LogEntry{Key: e.Key, Value: e.Value, At: e.At})
		}
		if page.Next == 0 {
			return out, nil
		}
		from = page.Next
	}
}

func (runtimeStore) Emit(event string, body any) error { return stdio.Emit(event, body) }

// recorder records into the hand-act log of the seat mesh-hand-acts alone (novox/hq ADR 0565 §11, issue 1953): the manifest
// invokes no verb of the controller and the SDK recorder has no fall back to its log.
var recorder = handacts.Recorder{Ask: stdio.Ask}

// HandAct records a change in the hand-act log (to-be 45 §7).
func (runtimeStore) HandAct(what, why, cause string) error {
	_, err := recorder.Record(handacts.Request{What: what, Why: why, Cause: cause})
	return err
}

// runtimeAsks is the operator channel through the runtime: asks published under this module's own name, the
// router's records of them read under that name alone (its `uses` of the seat, ADR 0259 §3), and the controller's
// hand-act log for a warrant acted on.
type runtimeAsks struct{}

func (runtimeAsks) Publish(subject string, body any, id string) error {
	return stdio.SeatPublish(subject, body, id)
}
func (runtimeAsks) Record(id string) (*stdio.StateEntry, error) { return stdio.SeatRecord("", id) }
func (runtimeAsks) Kinds() ([]stdio.KindHeld, error)            { return stdio.SeatKinds() }
func (runtimeAsks) Warranted(ask string) error {
	_, err := recorder.RecordWarrant(Module, ask)
	return err
}

func main() {
	c := &Config{Store: runtimeStore{}, Logf: logf, Asks: runtimeAsks{}, Forge: toolForge{ask: stdio.Ask},
		Delivery: toolDelivery{ask: stdio.Ask}, Vault: toolVault{ask: stdio.Ask, recorder: recorder}}
	moved := make(chan struct{}, 1)
	theViews = newOwnerViews(stdio.Ask, func() {
		select {
		case moved <- struct{}{}:
		default:
		}
	}, func(key string, v json.RawMessage) error {
		_, err := stdio.State(FindingsState).Put(key, v)
		return err
	},
		func(key string) error { return stdio.State(FindingsState).Delete(key) })
	theViews.Start(context.Background())
	go resolving(c, moved)
	go hearing(c)
	go hearingDelivery(c)
	if err := stdio.Serve("", Tools(c)); err != nil {
		logf("%v", err)
		os.Exit(1)
	}
}

// resolving resolves at start and again whenever a source moves, and every five minutes besides, so a change
// heard nowhere is still resolved. The runtime answers only once Serve runs, so the watches are asked again
// until they are taken.
func resolving(c *Config, moved chan struct{}) {
	// The owners' states reach `moved` through their views (ownerviews.go); the builder's builds, which has no tool yet,
	// is read when a want moves, and nothing else is watched.
	tick := time.NewTicker(5 * time.Minute)
	defer tick.Stop()
	time.Sleep(time.Second)
	for sweep := true; ; {
		if sweep {
			// A merge or a close of a pull request a value is held for, heard nowhere, is found here (ADR 0502).
			for _, s := range c.SweepPulls() {
				logf("[mesh-config] %s", s)
			}
		}
		sweep = false
		r, err := c.Reconcile()
		switch {
		case err != nil:
			logf("[mesh-config] the values are not all resolved: %v", err)
		case len(r.Moved) > 0:
			logf("[mesh-config] resolved: %d value(s) moved", len(r.Moved))
		}
		if r.Published != "" {
			logf("[mesh-config] published %s", r.Published)
		}
		select {
		case <-moved:
			time.Sleep(500 * time.Millisecond) // a burst of writes is one resolution
			select {
			case <-moved:
			default:
			}
		case <-tick.C:
			sweep = true
		}
	}
}

// hearing takes the router's word on this module's asks, asked again until the runtime takes the subscription, and
// reads the router's record of every waiting ask once a minute besides, so an answer heard nowhere is still taken.
func hearing(c *Config) {
	go func() {
		<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
		for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
			err := stdio.Subscribe(asks.Seat+".decided."+Module, func(e stdio.Envelope) error {
				var w asks.Warrant
				if json.Unmarshal(e.Body, &w) == nil {
					for _, s := range c.HeardEnd(w) {
						logf("[mesh-config] %s", s)
					}
				}
				return nil
			})
			if err == nil {
				logf("[mesh-config] hears the operator's answers")
				return
			}
			logf("[mesh-config] does not hear the operator's answers yet (%v); asking again in %s", err, wait)
			time.Sleep(wait)
		}
	}()
	for _, pattern := range []string{"gitea.pull.merged", "gitea.pull.closed", "gitea.forge.pull-request.closed"} {
		go func(pattern string) {
			<-stdio.Served() // subscribed as soon as the runtime serves, never after a guessed wait: an event sent meanwhile is heard
			for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
				err := stdio.Subscribe(pattern, func(e stdio.Envelope) error {
					var p PullEvent
					if json.Unmarshal(e.Body, &p) != nil {
						logf("[mesh-config] %s is not readable", pattern)
						return nil
					}
					for _, s := range c.HeardPull(p, pattern == "gitea.pull.merged") {
						logf("[mesh-config] %s", s)
					}
					return nil
				})
				if err == nil {
					return
				}
				logf("[mesh-config] does not hear %s yet (%v); asking again in %s", pattern, err, wait)
				time.Sleep(wait)
			}
		}(pattern)
	}
	tick := time.NewTicker(time.Minute)
	defer tick.Stop()
	for range tick.C {
		said, err := c.Decide()
		if err != nil {
			logf("[mesh-config] reading the operator's answers: %v", err)
		}
		for _, s := range said {
			logf("[mesh-config] %s", s)
		}
	}
}

// hearingDelivery takes the delivery parts' words that close a rename window or retire a value, each asked again until
// the runtime takes the subscription: the status view's `delivered`, the node-engine seat's `manifest-reads-changed`
// (a returning node-engine; design 32) and the indexer's `manifest-accepted`. At its start it judges the windows once,
// since an event said while this module was down is not said again.
func hearingDelivery(c *Config) {
	<-stdio.Served()
	for _, h := range []struct {
		pattern string
		handle  func(stdio.Envelope) []string
	}{
		{"mesh-delivery-view.delivered", func(stdio.Envelope) []string { return c.Windows("") }},
		{"node-engine.manifest-reads-changed", func(e stdio.Envelope) []string {
			var b struct {
				Node string `json:"node"`
			}
			_ = json.Unmarshal(e.Body, &b)
			if b.Node == "" {
				// The event's subject ends with the node's token (ADR 0618 as built): mesh.seat.node-engine.event.manifest-reads-changed.<node>.
				b.Node = e.Key[strings.LastIndex(e.Key, ".")+1:]
			}
			return c.Windows(b.Node)
		}},
		{"mesh-delivery-indexer.manifest-accepted", func(e stdio.Envelope) []string {
			var b struct {
				Module string `json:"module"`
			}
			if json.Unmarshal(e.Body, &b) != nil {
				return nil
			}
			return c.Retirements(b.Module)
		}},
	} {
		go func(pattern string, handle func(stdio.Envelope) []string) {
			for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) {
				err := stdio.Subscribe(pattern, func(e stdio.Envelope) error {
					for _, s := range handle(e) {
						logf("[mesh-config] %s", s)
					}
					return nil
				})
				if err == nil {
					return
				}
				logf("[mesh-config] does not hear %s yet (%v); asking again in %s", pattern, err, wait)
				time.Sleep(wait)
			}
		}(h.pattern, h.handle)
	}
	for _, s := range c.Windows("") {
		logf("[mesh-config] %s", s)
	}
}

// seatRename is one open rename as the seat's definition answers it: module, from, to, at and the nodes holding it open.
type seatRename struct {
	Module string    `json:"module"`
	From   string    `json:"from"`
	To     string    `json:"to"`
	At     time.Time `json:"at"`
	Open   []string  `json:"open"`
}

func seatRenames(rows []RenameRow) []seatRename {
	out := make([]seatRename, 0, len(rows))
	for _, r := range rows {
		out = append(out, seatRename{Module: r.Module, From: r.From, To: r.To, At: r.At, Open: r.Open})
	}
	return out
}

// Tools is the seat's verb and the module's own tools of the switch, as the runtime serves them.
func Tools(c *Config) []stdio.Tool {
	renames := stdio.Tool{Name: Seat + "." + RenamesVerb,
		Description: "The open renames of config values: for each, the module, the old and the new name, and the nodes " +
			"still holding the old name open (the rename window). The checker asks it before it lets a manifest add or " +
			"drop renamed-from (novox/hq design 32). Narrowed by module.",
		Input: map[string]any{"type": "object", "properties": map[string]any{
			"module": map[string]any{"type": "string", "description": "one module"}}},
		Run: func(a map[string]any) (any, error) {
			m, _ := a["module"].(string)
			rows, err := c.Renames(strings.TrimSpace(m), true)
			return map[string]any{"renames": seatRenames(rows)}, err
		}}
	return append(append(switchTools(c), viewTools(c, func() *ownerViews { return theViews })...), renames, stdio.Tool{
		Name: Seat + "." + Verb,
		Description: "Set, clear, list or read the history of config values (novox/hq ADR 0348). A value is set at " +
			"one level: mesh config (no module, no node) or node config (a node) for a predefined key, or a " +
			"module's own value on one node (module and node). Each set or clear needs why, is validated against " +
			"its declaration, names each module on each node whose value moves, and is kept in the history with " +
			"who and why. A trusted value is set only on the operator's Approve: " +
			"propose asks the operator, under mesh-config's own name on a channel that proves who answered, and sets " +
			"the value on Approve if what it was shown against still stands; proposals lists them, or one with " +
			"proposal. A proposal naming the open pull request that introduces its key (pull) is judged against what " +
			"that pull request declares and asked of the operator, and on Approve is held until the pull request " +
			"merges, then set; list shows the held values. pull-declares and held are the controller's. list and " +
			"history are narrowed by module, node and key.",
		Input: map[string]any{"type": "object", "properties": map[string]any{
			"action": map[string]any{"type": "string",
				"enum":        []string{"list", "set", "clear", "history", "propose", "proposals", "pull-declares", "held"},
				"description": "list when not given"},
			"key":    map[string]any{"type": "string", "description": "the config key"},
			"value":  map[string]any{"description": "for set: the value, of the key's declared type"},
			"module": map[string]any{"type": "string", "description": "a module: its own value, with node"},
			"node":   map[string]any{"type": "string", "description": "a node"},
			"why":    map[string]any{"type": "string", "description": "for set and clear: why, in a sentence"},
			"by": map[string]any{"type": "string",
				"description": "who gave the value at the controller's terminal: said by the controller alone"},
			"limit": map[string]any{"type": "number", "description": "for history: at most this many, newest"},
			"clear": map[string]any{"type": "boolean",
				"description": "for propose: propose clearing the value at its level instead of setting one"},
			"proposal": map[string]any{"type": "string", "description": "for proposals: one proposal's id, read whole"},
			"pull": map[string]any{"type": "string", "description": "owner/repository#number: for propose, the open " +
				"pull request that introduces the key, so the value is judged against what it declares and held until " +
				"it merges (a module's own value on one node); for pull-declares, the pull request declared for"},
			"head": map[string]any{"type": "string", "description": "for pull-declares: the commit the declarations are read at"},
			"declares": map[string]any{"type": "object", "description": "for pull-declares (the controller's): the " +
				"config declarations the head adds or changes against its base, by module and key"},
			"pulls": map[string]any{"type": "array", "description": "for held (the controller's): the pull requests " +
				"asked about, each {pull, head}", "items": map[string]any{"type": "object"}},
		}},
		Run: func(args map[string]any) (any, error) {
			call, err := callOf(args)
			if err != nil {
				return nil, err
			}
			call.Caller = stdio.CallerOf(args)
			return c.Do(call)
		},
	})
}

// callOf reads a call's arguments strictly: an argument it does not know is refused, not ignored.
func callOf(args map[string]any) (Call, error) {
	known := map[string]bool{"action": true, "key": true, "value": true, "module": true, "node": true, "why": true,
		"by": true, "limit": true, "clear": true, "proposal": true, "pull": true, "head": true, "declares": true,
		"pulls": true}
	var unknown []string
	for k := range args {
		if !known[k] {
			unknown = append(unknown, k)
		}
	}
	if len(unknown) > 0 {
		return Call{}, fmt.Errorf("config takes no %s; it takes action, key, value, module, node, why, by, limit, "+
			"clear, proposal, pull, head, declares and pulls",
			strings.Join(unknown, ", "))
	}
	str := func(k string) (string, error) {
		v, ok := args[k]
		if !ok || v == nil {
			return "", nil
		}
		s, ok := v.(string)
		if !ok {
			return "", fmt.Errorf("%s is a string", k)
		}
		return strings.TrimSpace(s), nil
	}
	var call Call
	var problems []error
	for k, into := range map[string]*string{"action": &call.Action, "key": &call.Key, "module": &call.Module,
		"node": &call.Node, "why": &call.Why, "by": &call.By, "proposal": &call.Proposal, "pull": &call.Pull,
		"head": &call.Head} {
		s, err := str(k)
		problems = append(problems, err)
		*into = s
	}
	if v, ok := args["limit"]; ok && v != nil {
		n, ok := v.(float64)
		if !ok || n < 0 || n != float64(int(n)) {
			problems = append(problems, errors.New("limit is a whole number"))
		}
		call.Limit = int(n)
	}
	if v, ok := args["clear"]; ok && v != nil {
		b, ok := v.(bool)
		if !ok {
			problems = append(problems, errors.New("clear is true or false"))
		}
		call.Clear = b
	}
	if v, ok := args["declares"]; ok && v != nil {
		m, ok := v.(map[string]any)
		if !ok {
			problems = append(problems, errors.New("declares is an object: a module to its keys' declarations"))
		}
		call.Declares = m
	}
	if v, ok := args["pulls"]; ok && v != nil {
		raw, _ := json.Marshal(v)
		if err := json.Unmarshal(raw, &call.Pulls); err != nil {
			problems = append(problems, errors.New("pulls is a list of {pull, head}"))
		}
	}
	if v, ok := args["value"]; ok {
		// The value as given, a JSON value: a number for a number, never its text; the validation judges it.
		call.Value = v
	}
	return call, errors.Join(problems...)
}