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

package grants

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

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

// 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.
//
// And the seat mesh-enrolment's keys (ADR 0587 §4, §5): a key at live composes enrol.<node>, and its end, by the holder or by
// the bus's expiry, is a change of the key that takes the user out; no event is needed beside the watch.
var Followed = []string{ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement, ReadEnrolments,
	ReadNodeLogins}

// 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 keeper'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
}

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 != 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) {
			// An account the vault minted.
			if IsAccountIssued(event) {
				f.Wake()
				return
			}
			// A condition event is heard from the keeper alone (ADR 0590 §3).
			if FromKeeper(event) && IsRootNotFree(body) {
				f.Wake()
			}
		})
		if err != nil {
			f.logf("the keeper'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()))
			}
		}
	}
}