mesh-delivery/modules/mesh-grants/internal/grants/follow.go

package grants

import (
	"context"
	"encoding/json"
	"strings"
	"sync"

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

// Watcher watches another part's state: changed hears each change after what stands, ended hears the watch end. A watch
// refused (a read not granted yet, a bucket not there) is an error, and is tried again at the next wake.
type Watcher interface {
	Watch(state string, changed func(key string, value []byte), ended func(error)) error
}

// Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved
// config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's
// wants (the manifest each machine is wanted to run), and the vault's hashes of the bus accounts.
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement, ReadLease, ReadTakeover}

// Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot
// open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect).
type Follower struct {
	Part  *Part
	Watch Watcher
	Logf  func(string, ...any)
	// Asked watches the part's own ask to compose now (a tool's compose-now), waking at each; nil watches none.
	Asked func(wake func()) error
	// Hear hears the controller seat's condition events and the vault's account-issued (Consumes) and hands each, by event
	// name, with its body, to handle; nil hears none. A condition of the kind root-not-free raised or cleared wakes a pass:
	// a machine's root-free word changed; and so does an account the vault minted (ADR 0561 §1), whose hash the list carries.
	Hear func(handle func(event string, body []byte)) error

	wake     chan struct{}
	once     sync.Once
	mu       sync.Mutex
	asked    bool
	heard_   bool
	rootSaid string
	watched  map[string]bool
	refused  map[string]string
	have     map[string]string
	// held and notAsserted are what the last pass held back and could not create, said again only when they change.
	held        string
	notAsserted string
	refusedMans string
	// leaseOn is whether the lease was handed away at its last change heard, once leaseSeen.
	leaseOn, leaseSeen bool
	// follows is the seat whose condition events the part hears, as the take-over key names it: the controller's until
	// the operator's take-over (an absent key, or one not readable, follows the controller).
	follows string
}

// Follows is the seat whose condition events the part hears now.
func (f *Follower) Follows() string {
	f.mu.Lock()
	defer f.mu.Unlock()
	if f.follows == "" {
		return "mesh-controller"
	}
	return f.follows
}

func (f *Follower) init() {
	f.once.Do(func() {
		f.wake = make(chan struct{}, 1)
		f.watched, f.refused, f.have = map[string]bool{}, map[string]string{}, map[string]string{}
	})
}

// Wake asks for one more composition: after a change, and after a reconnect.
func (f *Follower) Wake() {
	f.init()
	select {
	case f.wake <- struct{}{}:
	default:
	}
}

// Refused is every watch refused at the last try, with why.
func (f *Follower) Refused() map[string]string {
	f.init()
	f.mu.Lock()
	defer f.mu.Unlock()
	out := map[string]string{}
	for k, v := range f.refused {
		out[k] = v
	}
	return out
}

func (f *Follower) logf(format string, args ...any) {
	if f.Logf != nil {
		f.Logf(format, args...)
	}
}

// heard says whether a change composes again: every change but a have's, and of the haves a machine's first and the
// bus module's loaded list moving.
func (f *Follower) heard(state, key string, value []byte) bool {
	if state == ReadTakeover {
		// Which seat's condition events are heard: no pass, since nothing composed reads it.
		if key != health.TakeoverKey {
			return false
		}
		t, err := health.DecodeTakeover(value, len(value) > 0)
		if err != nil {
			f.logf("the take-over key cannot be read (%v): the controller's condition events are heard", err)
			t = health.Takeover{Follow: health.FollowController}
		}
		f.mu.Lock()
		was := f.follows
		f.follows = t.Seat()
		f.mu.Unlock()
		if was != t.Seat() {
			f.logf("root-free words are heard from %s's condition events (the take-over key follows %s)", t.Seat(), t.Follow)
		}
		return false
	}
	if state == ReadLease {
		// Renewed every few seconds by its holder: only the write that hands it away, or back, composes again (ADR 0561 §6).
		on, _ := buscompose.LeaseHandedAway(value, len(value) > 0, nil)
		f.mu.Lock()
		defer f.mu.Unlock()
		was, seen := f.leaseOn, f.leaseSeen
		f.leaseOn, f.leaseSeen = on, true
		return !seen || was != on
	}
	if state != ReadHave {
		return true
	}
	node, module, isNode, err := delivery.ParseHaveKey(key)
	if err != nil {
		return false
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	if isNode {
		if _, seen := f.have[node]; seen {
			return false
		}
		f.have[node] = ""
		return true
	}
	if module != BusModule {
		return false
	}
	var h struct {
		UserList json.RawMessage `json:"user-list"`
	}
	_ = json.Unmarshal(value, &h)
	was, seen := f.have[key]
	f.have[key] = string(h.UserList)
	return !seen || was != string(h.UserList)
}

// open opens every watch not open yet; one refused is said and left for the next wake.
func (f *Follower) open() {
	if f.Asked != nil && !f.asked {
		if err := f.Asked(f.Wake); err != nil {
			f.logf("a tool's ask to compose now cannot be heard yet (%v): tried again at the next change", err)
		} else {
			f.asked = true
		}
	}
	if f.Hear != nil && !f.heard_ {
		err := f.Hear(func(event string, body []byte) {
			if IsAccountIssued(event) {
				f.Wake()
				return
			}
			// A condition event of the seat the take-over key does not name is not heard (ADR 0564 §9).
			if from := HeardFrom(event); from != "" && from != f.Follows() {
				return
			}
			if IsRootNotFree(body) {
				f.Wake()
			}
		})
		if err != nil {
			f.logf("the controller's condition events cannot be heard yet (%v): tried again at the next change", err)
		} else {
			f.heard_ = true
		}
	}
	for _, state := range Followed {
		f.mu.Lock()
		open := f.watched[state]
		f.mu.Unlock()
		if open {
			continue
		}
		state := state
		err := f.Watch.Watch(state, func(key string, value []byte) {
			if f.heard(state, key, value) {
				f.Wake()
			}
		}, func(err error) {
			f.mu.Lock()
			f.watched[state] = false
			f.mu.Unlock()
			f.logf("the watch of %s ended (%v): opened again at the next change", state, err)
			f.Wake()
		})
		f.mu.Lock()
		if err != nil {
			f.refused[state] = err.Error()
			f.logf("%s cannot be watched yet (%v): composed without it, and tried again at the next change or reconnect", state, err)
		} else {
			f.watched[state] = true
			delete(f.refused, state)
		}
		f.mu.Unlock()
	}
}

// said logs every membership the pass held back, each with what it would lose (issue 1945), and every declared stream
// the pass could not create, each with why (issue 1790), when the set changes from the last pass's; and every stream it
// created, one line per module.
func (f *Follower) said(res Result) {
	if refused := strings.Join(res.Refused, "\n"); refused != f.refusedMans {
		f.refusedMans = refused
		if len(res.Refused) == 0 {
			f.logf("every manifest the indexer holds is read")
		} else {
			f.logf("%d manifests cannot be read, so their modules count as placed nowhere (grants_membership with no node and module lists them):",
				len(res.Refused))
			for _, r := range res.Refused {
				f.logf("refused %s", r)
			}
		}
	}
	if held := strings.Join(res.NotPublished, "\n"); held != f.held {
		f.held = held
		if len(res.NotPublished) == 0 {
			f.logf("no membership held back")
		} else {
			f.logf("%d memberships not published, tried again at the next event (grants_membership with no node and module lists them):",
				len(res.NotPublished))
			for _, n := range res.NotPublished {
				f.logf("held %s", n)
			}
		}
	}
	if said := strings.Join(res.RootNotFree, "\n"); said != f.rootSaid {
		f.rootSaid = said
		if len(res.RootNotFree) == 0 {
			f.logf("every machine a composition needs a root-free word for is free")
		}
		for _, n := range res.RootNotFree {
			f.logf("root-free: %s (a verified sender is composed as from no root-free machine there)", n)
		}
	}
	for _, a := range res.Asserted {
		f.logf("asserted %s", a)
	}
	if not := strings.Join(res.NotAsserted, "\n"); not != f.notAsserted {
		f.notAsserted = not
		for _, n := range res.NotAsserted {
			f.logf("not asserted %s", n)
		}
	}
}

// Run composes at start and after every change until ctx ends. It returns only when ctx ends.
func (f *Follower) Run(ctx context.Context) {
	f.init()
	f.Wake()
	for {
		select {
		case <-ctx.Done():
			return
		case <-f.wake:
			f.open()
			res := f.Part.Pass()
			for _, e := range res.Errors {
				f.logf("%s", e)
			}
			if len(res.Published) > 0 {
				f.logf("published %d memberships: %v", len(res.Published), res.Published)
			}
			f.said(res)
			if res.Written {
				f.logf("the shadow list moved to revision %d: %d users, %d differing from the loaded list",
					res.Revision, res.UserList.Users, len(res.UserList.Compare.Differing()))
			}
		}
	}
}