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

package grants

import (
	"context"
	"encoding/json"
	"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).
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants}

// 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

	wake    chan struct{}
	once    sync.Once
	mu      sync.Mutex
	asked   bool
	watched map[string]bool
	refused map[string]string
	have    map[string]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
		}
	}
	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()
	}
}

// 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)
			}
			if len(res.NotPublished) > 0 {
				f.logf("%d memberships not published, tried again at the next event; the first: %s", len(res.NotPublished), res.NotPublished[0])
			}
			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()))
			}
		}
	}
}