mesh-tools/node-tools/internal/firstview/firstview.go

// Package firstview is the tool runner's picture of what it reads of the assigner and the catalog half: the assigner's
// assignments and seat-holders (novox/hq ADR 0615 §1, §3; ADR 0621 §2; issue 1966 TLS-5), and which modules exist with the
// manifest each entry carries, followed per watched branch through meshcatalogclient's branch follower on this process's own
// consumer (design 47 G §196, §594, §614; lead answers A-5, A-27, A-103), never the indexer's state `modules` or `manifests`.
//
// **It never reads the owners' state.** At start it asks each owner's first-view tool until answered (the owner answers in a
// runtime of its own, never through this one: ADR 0621 §1), then follows the owner's views family, one event per changed key;
// a gap in the revisions, or an owner whose state was made again, asks the first view once more, said in the log. Past two
// minutes without an answer it writes a finding through its held finding writer (design 47 G A-27, A-28), which keeps it in its
// own state `findings` and says it; the answer clears it.
package firstview

import (
	"context"
	"encoding/json"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

	condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
	catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/keys"
	"git.novox.be/novox/mesh-sdk/go/views"

	"github.com/novox/mesh-tools/node-tools/internal/bus"
)

// The owner and its first-view tools (the tools' addresses are `<owner>.<tool>`).
const Assigner = "mesh-delivery-assigner"

// The names the console reads the assigner's two states by (`<owner>.<state>`), as the manifest's `consumes` names their views
// families (no seat's client names a holder, novox/hq ADR 0625 §4; TestTheFirstViewsAreTheManifestsViews).
const (
	AssignmentsRead = Assigner + ".assignments"
	SeatHoldersRead = Assigner + ".seat-holders"
)

// Spec is one first view the runner holds.
type Spec struct{ Owner, Tool, State string }

// Specs are the two owner states asked of an owner's tool; the catalog half's views are adopted by the branch follower.
var Specs = []Spec{
	{Assigner, "assignments", "assignments"},
	{Assigner, "seat-holders", "seat-holders"},
}

// Name is the name a state is read by, `<owner>.<state>`: AssignmentsRead and SeatHoldersRead.
func (s Spec) Name() string { return s.Owner + "." + s.State }

// Views is the first views: the assigner's two, and the catalog half's per watched branch.
type Views struct {
	logf      func(string, ...any)
	mu        sync.Mutex
	followers map[string]*views.Follower // by `<owner>.<state>`
	byState   map[string]*views.Follower // by the state an event names, the owners' views families
	adopted   map[string]*adoption       // by `<owner>.<state>`, the views the branch follower adopted
	hearers   map[string][]*adoption     // event key `<seat>.<event>` -> the adopted views that hear it
	branches  *meshcatalogclient.Branches
	ctx       context.Context
	cancel    context.CancelFunc
	wg        sync.WaitGroup
	deps      Deps
}

// FindingsState is the runner's own state the findings are written in.
const FindingsState = "findings"

// Deps is what the views reach the bus through: Start makes them from the runner's connection, a test from fakes.
type Deps struct {
	// Module is the runner's own name, the observer of its findings.
	Module string
	// Ask calls an owner's tool, `<owner>.<tool>`.
	Ask func(address string, body any) (json.RawMessage, error)
	// Consume hands each event of the owners' views families to deliver, until the stop is called.
	Consume func(deliver func(bus.Envelope) error) (stop func(), err error)
	// Findings is the runner's held finding writer (views.HeldWriter): a follower's unanswered first view is written and cleared
	// through it with the contract's Finding (design 47 G A-28). Nil says it in the log only.
	Findings FindingWriter
	// StartFindings starts that writer, and its stop ends it; nil starts none.
	StartFindings func(ctx context.Context) (stop func(), err error)
	// Node is the machine the runner is on, named in its findings' subject.
	Node string
	// Bound and Wait are the follower's (zero: two minutes, five seconds).
	Bound, Wait time.Duration
}

// FindingWriter is the writer the runner's findings go through: views.HeldWriter over the contract's Finding.
type FindingWriter interface {
	Write(ctx context.Context, finding condseat.Finding) error
	Clear(ctx context.Context, scope, id, kind, why string) error
}

// Start asks the first views, each until answered, and follows the owners' views families on the runner's own
// consumers. It returns at once: the picture fills as the owners answer.
func Start(parent context.Context, conn *bus.Conn, logf func(string, ...any)) (*Views, func()) {
	return StartWith(parent, DepsOf(conn, logf), logf)
}

// DepsOf is the views' reach through the runner's own connection: its tool calls, its consumer and its held finding writer, whose
// state adapter, JetStream and node are this connection's, in this one place (design 47 G A-16, A-27).
func DepsOf(conn *bus.Conn, logf func(string, ...any)) Deps {
	me := conn.Module()
	d := Deps{Module: me, Node: conn.Node(),
		Ask: func(address string, body any) (json.RawMessage, error) {
			got, err := conn.Ask(address, body, "")
			return got.Result, err
		},
		Consume: func(deliver func(bus.Envelope) error) (func(), error) { return conn.ConsumeAs(me, deliver) },
	}
	js, err := conn.JetStream()
	if err != nil {
		logf("[mesh-tools] the runner's findings are not written (%v): the first views' findings are in the log only", err)
		return d
	}
	w, err := views.NewHeldWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{
		Asked: condseat.SubjectFindingsAsked, Module: me, Node: conn.Node(), State: findingsState{conn: conn, module: me}, JS: js, Logf: logf,
		Kinds: []string{views.UnansweredKind}})
	if err != nil {
		logf("[mesh-tools] the runner's findings are not written (%v): the first views' findings are in the log only", err)
		return d
	}
	d.Findings = w
	d.StartFindings = func(ctx context.Context) (func(), error) { return w.Close, w.Start(ctx) }
	return d
}

// findingsState is the runner's own state `findings` (views.FindingState) over its connection.
type findingsState struct {
	conn   *bus.Conn
	module string
}

func (s findingsState) Get(key string) (*stdio.StateEntry, error) {
	e, err := s.conn.StateGet(s.module, FindingsState, key)
	if err != nil || e == nil {
		return nil, err
	}
	return &stdio.StateEntry{Key: e.Key, Value: e.Value, Revision: e.Revision, At: e.At}, nil
}

func (s findingsState) Put(key string, value any) (uint64, error) {
	raw, err := json.Marshal(value)
	if err != nil {
		return 0, err
	}
	return s.conn.StatePut(s.module, FindingsState, key, raw)
}

func (s findingsState) Delete(key string) error {
	return s.conn.StateDelete(s.module, FindingsState, key)
}

func (s findingsState) Keys() ([]string, error) { return s.conn.StateKeys(s.module, FindingsState) }

// StartWith is Start over deps.
func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*Views, func()) {
	ctx, cancel := context.WithCancel(parent)
	v := &Views{logf: logf, ctx: ctx, cancel: cancel, followers: map[string]*views.Follower{}, byState: map[string]*views.Follower{},
		adopted: map[string]*adoption{}, hearers: map[string][]*adoption{}, deps: d}
	for _, s := range Specs {
		s := s
		call := func(tool string, body any) (json.RawMessage, error) { return d.Ask(s.Owner+"."+tool, body) }
		f := &views.Follower{Owner: s.Owner, Tool: s.Tool, State: s.State, Ask: views.AskVia(call, s.Tool), Logf: logf,
			Bound: d.Bound, Wait: d.Wait,
			Raise: func(st views.Status) {
				v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, true)
			},
			Clear: func(st views.Status) {
				v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
			}}
		v.followers[s.Name()] = f
		v.byState[s.State] = f
		v.wg.Add(1)
		go func() { defer v.wg.Done(); _ = f.Start(ctx) }()
	}
	stopFindings := func() {}
	if d.StartFindings != nil {
		if stop, err := d.StartFindings(ctx); err != nil {
			logf("[mesh-tools] the runner's finding writer does not start (%v): the first views' findings are in the log only", err)
		} else {
			stopFindings = stop
		}
	}
	// The catalog half is followed by the shared branch follower on this process's own wiring (design 47 G §614, A-103): the
	// registrations, the gate's source and a view per watched branch, each heard on this consumer.
	branches, err := meshcatalogclient.NewBranches(wiring{v}, func(key string, body any) (json.RawMessage, error) { return d.Ask(key, body) })
	if err != nil {
		logf("[mesh-tools] the catalog half is not followed (%v): the console cannot say which modules exist", err)
	} else {
		branches.Logf = func(format string, args ...any) { logf("[mesh-tools] "+format, args...) }
		v.branches = branches
	}
	stop, err := d.Consume(v.hear)
	if err != nil {
		logf("[mesh-tools] the owners' views are not followed (%v): the pictures are the first views alone until the runner starts again", err)
		stop = func() {}
	}
	return v, func() { stop(); cancel(); v.wg.Wait(); stopFindings() }
}

// hear applies one views event; a gap asks the first view once more.
func (v *Views) hear(env bus.Envelope) error {
	v.mu.Lock()
	hs := append([]*adoption(nil), v.hearers[env.Key]...)
	v.mu.Unlock()
	if len(hs) > 0 { // an event of the catalog half or the gate's source, heard by the views the branch follower adopted
		for _, a := range hs {
			if err := a.hear(v.ctx, env.Body); err != nil {
				v.logf("[mesh-tools] %v", err)
			}
		}
		return nil
	}
	ev, err := views.DecodeEvent(env.Body)
	if err != nil {
		return nil // not a views event: nothing to hold, nothing to retry
	}
	f := v.byState[ev.State]
	if f == nil || f.Owner != ev.Owner {
		return nil
	}
	if gap, why := f.Apply(ev); gap {
		v.logf("[mesh-tools] a gap in %s's %s (%s): its first view %s is asked once more", f.Owner, f.State, why, f.Tool)
		go func() { _ = f.Again(v.ctx, "a gap: "+why) }()
	}
	return nil
}

// finding writes or clears the runner's finding of a first view unanswered past its bound, through its writer (design 47 G A-28).
func (v *Views) finding(f condseat.Finding, tool string, raised bool) {
	if v.deps.Findings == nil {
		v.logf("[mesh-tools] %s", f.Summary)
		return
	}
	var err error
	if raised {
		err = v.deps.Findings.Write(context.Background(), f)
	} else {
		err = v.deps.Findings.Clear(context.Background(), f.Scope, f.ID, f.Kind, "the owner answered")
	}
	if err != nil {
		v.logf("[mesh-tools] the finding of the first view %s is not kept: %v", tool, err)
	}
}

// State is the picture of the state spec names (`<owner>.<state>`), every current value by key; an error says the owner has
// not answered yet, so nothing is claimed of it.
func (v *Views) State(name string) (map[string][]byte, error) {
	v.mu.Lock()
	f := v.followers[name]
	v.mu.Unlock()
	if f == nil {
		return nil, fmt.Errorf("the tool runner holds no first view of %s", name)
	}
	select {
	case <-f.Ready():
	default:
		return nil, fmt.Errorf("%s's first view %s has not been answered yet", f.Owner, f.Tool)
	}
	out := map[string][]byte{}
	for k, raw := range f.Values() {
		out[k] = raw
	}
	return out, nil
}

// Entry is one value of the picture, nil when the owner holds none.
func (v *Views) Entry(name, key string) ([]byte, error) {
	all, err := v.State(name)
	if err != nil {
		return nil, err
	}
	return all[key], nil
}

// Status is each first view as it stands, by `<owner>.<state>`, sorted.
func (v *Views) Status() []views.Status {
	var out []views.Status
	for _, f := range v.all() {
		out = append(out, f.Status())
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Owner+out[i].State < out[j].Owner+out[j].State })
	return out
}

// Entries are the entries by fingerprint the catalog half's follower keeps, with their state and since when (design 47 G A-103).
func (v *Views) Entries() any {
	if v.branches == nil {
		return []meshcatalogclient.EntryStatus{}
	}
	return v.branches.EntryStatuses()
}

// all is every first view held now: the owners', then the adopted ones.
func (v *Views) all() []*views.Follower {
	v.mu.Lock()
	defer v.mu.Unlock()
	out := make([]*views.Follower, 0, len(v.followers))
	for _, f := range v.followers {
		out = append(out, f)
	}
	return out
}

// Again asks every first view once more, by hand, for why.
func (v *Views) Again(ctx context.Context, why string) []error {
	var errs []error
	for _, f := range v.all() {
		if err := f.Again(ctx, why); err != nil {
			errs = append(errs, err)
		}
	}
	return errs
}

// unansweredFinding is the finding of a first view unanswered past its bound, in the contract's form (design 47 G A-28): about the
// runner's module, a warning, its id the view's name escaped as a key's name is.
func unansweredFinding(module, node string, st views.Status, at time.Time) condseat.Finding {
	return condseat.Finding{Kind: views.UnansweredKind, Scope: scopeModule, ID: keys.EscapeName(module + "." + st.Owner + "." + st.State),
		Severity: condseat.FindingSeverityWarning, Source: "views", Observer: module,
		Subject: condseat.Subject{Scope: scopeModule, ID: module, Machine: node},
		Summary: fmt.Sprintf("%s has no first view of %s's %s: %s has not answered since %s (%s)", module, st.Owner, st.State, st.Tool,
			st.Asked.Format(time.RFC3339), strings.TrimSpace(st.Error)),
		Headline: fmt.Sprintf("%s waits for %s", module, st.Owner), At: at.UTC()}
}

// scopeModule is the scope of a finding about a module.
const scopeModule = "module"

// viewWaitsFinding is the finding of a view waited for past views.WaitsBound, in the contract's form: about the runner's module,
// a warning, its id the view's name escaped as a key's name is.
func viewWaitsFinding(module, node string, st views.Status, at time.Time) condseat.Finding {
	return condseat.Finding{Kind: views.ViewWaitsKind, Scope: scopeModule, ID: keys.EscapeName(module + "." + st.Owner + "." + st.State),
		Severity: condseat.FindingSeverityWarning, Source: "views", Observer: module,
		Subject: condseat.Subject{Scope: scopeModule, ID: module, Machine: node},
		Summary: fmt.Sprintf("%s waits for the view %s.%s at revision %s since %s and acts on nothing new until it holds it; it holds %s",
			module, st.Owner, st.State, st.WaitsFor, st.WaitingSince.UTC().Format(time.RFC3339), orNone(st.View)),
		Headline: fmt.Sprintf("%s waits for a view of %s", module, st.Owner), At: at.UTC()}
}

func orNone(s string) string {
	if s == "" {
		return "none"
	}
	return s
}

// adoption is a view the branch follower put on this process's consumer: its follower, the event it hears and its handler.
type adoption struct {
	name, key string
	f         *views.Follower
	hear      func(context.Context, []byte) error
	cancel    context.CancelFunc
}

// wiring is the branch follower's wiring over this process's own consumer (meshcatalogclient.Wiring; design 47 G §614, A-103):
// an adopted view is held with the owners', takes its first view at once, hears its seat's event on the runner's consumer,
// writes its findings through the runner's held writer, and a dropped one clears them.
type wiring struct{ v *Views }

func (w wiring) Adopt(f *views.Follower, seat, event string, hear func(context.Context, []byte) error) error {
	v := w.v
	if f == nil || f.Owner == "" || f.State == "" || seat == "" || event == "" || hear == nil {
		return fmt.Errorf("the tool runner adopts a view that names its owner, state, seat's event and handler")
	}
	name := f.Owner + "." + f.State
	key := seat + "." + event
	v.mu.Lock()
	if v.followers[name] != nil {
		v.mu.Unlock()
		return fmt.Errorf("%s is followed already", name)
	}
	if f.Logf == nil {
		f.Logf = v.logf
	}
	if f.Bound == 0 {
		f.Bound = v.deps.Bound
	}
	if f.Wait == 0 {
		f.Wait = v.deps.Wait
	}
	held := func() bool { v.mu.Lock(); defer v.mu.Unlock(); return v.followers[name] == f }
	f.Raise = func(st views.Status) {
		if held() {
			v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, true)
		}
	}
	f.Clear = func(st views.Status) {
		v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
	}
	f.RaiseWaits = func(st views.Status) {
		if held() {
			v.finding(viewWaitsFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, true)
		}
	}
	f.ClearWaits = func(st views.Status) {
		v.finding(viewWaitsFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
	}
	ctx, cancel := context.WithCancel(v.ctx)
	a := &adoption{name: name, key: key, f: f, hear: hear, cancel: cancel}
	v.followers[name] = f
	v.adopted[name] = a
	v.hearers[key] = append(v.hearers[key], a)
	v.mu.Unlock()
	v.wg.Add(1)
	go func() { defer v.wg.Done(); _ = f.Start(ctx) }()
	return nil
}

func (w wiring) Drop(owner, state string) error {
	v := w.v
	name := owner + "." + state
	v.mu.Lock()
	a := v.adopted[name]
	if a == nil {
		v.mu.Unlock()
		return fmt.Errorf("%s is not a view this tool runner adopted", name)
	}
	delete(v.followers, name)
	delete(v.adopted, name)
	v.hearers[a.key] = slices.DeleteFunc(v.hearers[a.key], func(o *adoption) bool { return o == a })
	v.mu.Unlock()
	a.cancel()
	st := a.f.Status()
	if st.Raised {
		v.finding(unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
	}
	if st.WaitsRaised {
		v.finding(viewWaitsFinding(v.deps.Module, v.deps.Node, st, time.Now()), st.Tool, false)
	}
	return nil
}

// Halves is the catalog half of each followed branch, by branch: its revision, with the seat definitions it was judged
// with, and its entries; mesh_wait checks an event name's emitter against them (design 34 every-event §14). An error says
// the half has no first view yet, never that no module exists.
func (v *Views) Halves() (map[string]meshcatalogclient.Half, error) {
	if v.branches == nil {
		return nil, fmt.Errorf("the tool runner does not follow the catalog half")
	}
	if err := v.branches.Ready(); err != nil {
		return nil, err
	}
	out := map[string]meshcatalogclient.Half{}
	for _, branch := range v.branches.Followed() {
		if c, ok := v.branches.Catalog(branch); ok {
			out[branch], _ = c.View() // a branch at none-yet is ready and empty (§614)
		}
	}
	return out, nil
}

// Modules is every module the catalog half holds, by name, with its entry: the entries of the watched branches' newest trunk
// revisions, the branch followed first naming a module more than one hold (design 47 G §196, §594). An error says the half has
// no first view yet, never that no module exists.
func (v *Views) Modules() (map[string]catalogseat.CatalogEntry, error) {
	if v.branches == nil {
		return nil, fmt.Errorf("the tool runner does not follow the catalog half")
	}
	if err := v.branches.Ready(); err != nil {
		return nil, err
	}
	out := map[string]catalogseat.CatalogEntry{}
	for _, branch := range v.branches.Followed() {
		c, ok := v.branches.Catalog(branch)
		if !ok {
			continue
		}
		half, _ := c.View() // a branch at none-yet is ready and empty (§614)
		for m, e := range half.Entries {
			if _, taken := out[m]; !taken {
				out[m] = e
			}
		}
	}
	return out, nil
}