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

package grants

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

	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
	occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
	oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
	"git.novox.be/novox/mesh-sdk/go/membership"
	"git.novox.be/novox/mesh-sdk/go/views"
)

// **The first view of the bindings** (design 25 bindings-first-view). The bus module's events were the binding's only
// source, so an event the part never heard lost the binding for good. The part now asks the seat mesh-broker's `list` once
// at each start (§3) and again when a person asks (grants_first_view_again, state `bindings`), never on a timer, and takes
// each binding the answer carries into its kept bindings (§4). The live events stay the way a binding changes between starts.

// The first view of the bindings: its name for grants_first_view_again, the seat and verb it asks, and its conditions.
const (
	FirstViewBindings = "bindings"
	BrokerSeat        = meshbroker.Seat
	BrokerListVerb    = meshbroker.VerbList
	// CallBrokerList is the seat's verb as the manifest's invokes names it.
	CallBrokerList = "seat:" + meshbroker.Seat + "." + meshbroker.VerbList
	// BindingUnlisted is the condition of a binding the part holds that the bus module no longer lists (§4), one per binding.
	BindingUnlisted = "mesh-grants.binding-unlisted"
)

// BindingsStatus is what the part's tools say of the first view of the bindings (§6), kept beside the other first views.
type BindingsStatus struct {
	Seat     string    `json:"seat"`
	Verb     string    `json:"verb"`
	Asked    time.Time `json:"asked,omitzero"`
	Answered time.Time `json:"answered,omitzero"`
	// Listed is how many entries the last answer listed; Written and Rewritten how many bindings it wrote and rewrote.
	Listed    int `json:"listed"`
	Written   int `json:"written"`
	Rewritten int `json:"rewritten"`
	// Composed is how many of the bindings the last answer wrote or rewrote were composed from its own fields, and Unbound
	// the entries it left unbound, `<module>/<local name> (<storage type>)`: both only for an answer no entry of which
	// carries `binding` (§9).
	Composed int      `json:"composed"`
	Unbound  []string `json:"unbound,omitempty"`
	// Unlisted are the bindings the part holds that the last answer did not list, `<module>/<local name>`.
	Unlisted []string `json:"unlisted,omitempty"`
	Asks     int      `json:"asks"`
	ByHand   int      `json:"by-hand"`
	// Raised is whether the condition first-view-unanswered is raised for it.
	Raised bool   `json:"raised"`
	Why    string `json:"why,omitempty"`
	Error  string `json:"error,omitempty"`
}

// BindingsView is the first view of the bindings.
type BindingsView struct {
	// Ask calls the seat's list and answers its reply as the tool's own text (the process's connection).
	Ask func(ctx context.Context) (json.RawMessage, error)
	// Say says a condition; nil says none. Changed wakes a pass; Logf logs; Now is the clock.
	Say     func(oc.Condition) error
	Changed func()
	Logf    func(string, ...any)
	Now     func() time.Time
	// Pause waits d between two asks and answers false once ctx ended; nil waits on the context's bound.
	Pause func(ctx context.Context, d time.Duration) bool
	// Bound is how long it asks before it raises its condition (two minutes: views.Bound when zero); Wait the first wait
	// between two asks that failed (ten seconds when zero), doubling to MaxWait (five minutes when zero).
	Bound   time.Duration
	Wait    time.Duration
	MaxWait time.Duration

	mu      sync.Mutex
	st      BindingsStatus
	asking  bool
	holding bool
	held    []heldEvent
	touched map[string]bool
	raised  map[string]bool // the unlisted conditions raised, by `<module>/<local name>`
}

type heldEvent struct {
	seq   uint64
	event string
	body  []byte
	ack   func()
}

func (v *BindingsView) now() time.Time {
	if v.Now != nil {
		return v.Now().UTC()
	}
	return time.Now().UTC()
}

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

func (v *BindingsView) wake() {
	if v.Changed != nil {
		v.Changed()
	}
}

func (v *BindingsView) say(c oc.Condition) {
	if v.Say == nil {
		return
	}
	if err := v.Say(c); err != nil {
		v.logf("the condition %s could not be said: %v", c.Key, err)
	}
}

// Status is the first view's status now.
func (v *BindingsView) Status() BindingsStatus {
	v.mu.Lock()
	defer v.mu.Unlock()
	return v.statusLocked()
}

func (v *BindingsView) statusLocked() BindingsStatus {
	s := v.st
	s.Seat, s.Verb = BrokerSeat, BrokerListVerb
	s.Unlisted = slices.Sorted(func(yield func(string) bool) {
		for k := range v.raised {
			if !yield(k) {
				return
			}
		}
	})
	return s
}

func localKey(module, local string) string { return module + "/" + local }

// Hold keeps an event of the seat heard before the first view is applied (§3), and says whether it did. A held event is not
// acknowledged: ack is called only when the event is applied after the answer (or past the bound), so a restart finds it
// delivered again. seq is the event's sequence on its stream; one delivered again while held replaces its ack and keeps its place.
func (v *BindingsView) Hold(seq uint64, event string, body []byte, ack func()) bool {
	v.mu.Lock()
	defer v.mu.Unlock()
	if !v.holding {
		return false
	}
	for i := range v.held {
		if seq != 0 && v.held[i].seq == seq {
			v.held[i].ack = ack
			return true
		}
	}
	v.held = append(v.held, heldEvent{seq, event, slices.Clone(body), ack})
	return true
}

// touch notes that an event named the entry since the last ask was sent, so an answer never writes over it (§4); a dropped
// one also clears the entry's unlisted condition.
func (v *BindingsView) touch(module, local string, dropped bool) {
	v.mu.Lock()
	if v.touched == nil {
		v.touched = map[string]bool{}
	}
	v.touched[localKey(module, local)] = true
	clear := dropped && v.raised[localKey(module, local)]
	if clear {
		delete(v.raised, localKey(module, local))
	}
	v.mu.Unlock()
	if clear {
		v.say(unlistedCondition(module, local, false))
	}
}

// Run asks the first view and answers when it is answered (or ctx ends); Start runs it on a goroutine of its own.
func (v *BindingsView) Run(ctx context.Context, p *Part, why string, byHand bool) error {
	return v.askUntil(ctx, p, why, byHand)
}

// Start asks the first view once at the part's start, on a goroutine of its own, holding the seat's events until the answer
// is applied or the bound passes (§3). The unlisted conditions raised before the start are read from the part's kept status.
func (v *BindingsView) Start(ctx context.Context, p *Part) {
	v.mu.Lock()
	v.holding = true
	v.mu.Unlock()
	v.seed(p)
	go func() { _ = v.askUntil(ctx, p, "start", false) }()
}

// Again asks the first view once more by hand (grants_first_view_again), on a goroutine of its own.
func (v *BindingsView) Again(ctx context.Context, p *Part, why string) {
	go func() { _ = v.askUntil(ctx, p, "by hand: "+why, true) }()
}

// seed reads the unlisted bindings the last run kept, so a later answer that lists them clears their conditions.
func (v *BindingsView) seed(p *Part) {
	raw, _, found, err := p.Store.Get(StateUserList, KeyFirstViews)
	if err != nil || !found {
		return
	}
	var st FirstViewsState
	if json.Unmarshal(raw, &st) != nil || st.Bindings == nil {
		return
	}
	v.mu.Lock()
	defer v.mu.Unlock()
	v.raised = map[string]bool{}
	for _, k := range st.Bindings.Unlisted {
		v.raised[k] = true
	}
}

func (v *BindingsView) pause(ctx context.Context, d time.Duration) bool {
	if v.Pause != nil {
		return v.Pause(ctx, d)
	}
	t, cancel := context.WithTimeout(ctx, d)
	<-t.Done()
	cancel()
	return ctx.Err() == nil
}

// askUntil asks until the bus module answers or ctx ends: from ten seconds, doubling, at most every five minutes (§5).
// Past the bound it raises first-view-unanswered, releases the events it holds, and keeps asking; the answer clears it.
func (v *BindingsView) askUntil(ctx context.Context, p *Part, why string, byHand bool) error {
	v.mu.Lock()
	if v.asking {
		v.mu.Unlock()
		return nil // one ask at a time
	}
	v.asking = true
	v.st.Why = why
	if byHand {
		v.st.ByHand++
	}
	v.mu.Unlock()
	defer func() {
		v.mu.Lock()
		v.asking = false
		v.mu.Unlock()
	}()
	bound, wait, most := v.Bound, v.Wait, v.MaxWait
	if bound == 0 {
		bound = views.Bound
	}
	if wait == 0 {
		wait = 10 * time.Second
	}
	if most == 0 {
		most = 5 * time.Minute
	}
	first := v.now()
	for {
		v.mu.Lock()
		v.touched = map[string]bool{} // what an event names from here on is newer than this ask
		v.st.Asked = v.now()
		v.st.Asks++
		v.mu.Unlock()
		raw, err := v.Ask(ctx)
		if err == nil {
			err = v.apply(p, raw)
		}
		if err == nil {
			return nil
		}
		v.mu.Lock()
		v.st.Error = err.Error()
		raise := !v.st.Raised && v.now().Sub(first) >= bound
		if raise {
			v.st.Raised = true
		}
		v.mu.Unlock()
		v.logf("the first view %s of the bindings is not answered (%v): asked again in %s", CallBrokerList, err, wait)
		if raise {
			v.say(bindingsUnansweredCondition(true, v.Status()))
			v.release(p)
		}
		v.wake()
		d := wait // no pause runs past the bound: the ask at the bound raises and releases (§3, §5)
		v.mu.Lock()
		raised := v.st.Raised
		v.mu.Unlock()
		if left := bound - v.now().Sub(first); !raised && left < d {
			d = max(left, 0)
		}
		if !v.pause(ctx, d) {
			return ctx.Err()
		}
		if wait *= 2; wait > most {
			wait = most
		}
	}
}

// release takes the events held, in the order heard, after the answer or past the bound.
func (v *BindingsView) release(p *Part) {
	v.mu.Lock()
	held := v.held
	v.held, v.holding = nil, false
	v.mu.Unlock()
	changed := false
	for _, e := range held {
		if c, err := p.hearBroker(e.event, e.body, p.Brokers()); err != nil {
			v.logf("%v: not applied, left unacknowledged to be delivered again", err)
			continue
		} else if c {
			changed = true
		}
		if e.ack != nil {
			e.ack()
		}
	}
	if changed {
		v.wake()
	}
}

// apply takes an answer of list into the kept bindings (§4) and then the events held (§3).
func (v *BindingsView) apply(p *Part, raw json.RawMessage) error {
	var answer meshbroker.ListAnswer
	if err := json.Unmarshal(raw, &answer); err != nil {
		return fmt.Errorf("%s answered what is not a list: %w", CallBrokerList, err)
	}
	entries, composed, unbound := v.compose(answer.Entries)
	p.bindMu.Lock()
	written, rewritten, fromFields, err := v.take(p, entries, composed)
	p.bindMu.Unlock()
	if err != nil {
		return err
	}
	v.mu.Lock()
	wasRaised := v.st.Raised
	v.st.Raised, v.st.Error = false, ""
	v.st.Answered = v.now()
	v.st.Listed = len(answer.Entries)
	v.st.Written, v.st.Rewritten = written, rewritten
	v.st.Composed, v.st.Unbound = fromFields, unbound
	status := v.statusLocked()
	v.mu.Unlock()
	if wasRaised {
		v.say(bindingsUnansweredCondition(false, status))
	}
	v.release(p)
	v.wake()
	return nil
}

// compose gives the entries of an answer that carries no `binding` on any entry (a bus module's build from before the
// field) the binding its own fields name whole (design 25 bindings-first-view §9): for a key-value entry or an object-store
// entry whose settings hold the bus object its bus name names. It answers the entries with those bindings, the entries
// composed (`<module>/<local name>`) and the ones left unbound, and says once what it did. An answer where any entry carries
// `binding` is applied by §4 alone: it answers the entries as they are.
func (v *BindingsView) compose(entries []meshbroker.Held) ([]meshbroker.Held, map[string]bool, []string) {
	if len(entries) == 0 || slices.ContainsFunc(entries, func(e meshbroker.Held) bool { return e.Binding != nil }) {
		return entries, nil, nil
	}
	out := slices.Clone(entries)
	composed := map[string]bool{}
	var unbound []string
	for i, e := range out {
		if b, ok := composeBinding(e); ok {
			out[i].Binding = b
			composed[localKey(e.Module, e.LocalName)] = true
			continue
		}
		unbound = append(unbound, fmt.Sprintf("%s (%s)", localKey(e.Module, e.LocalName), e.StorageType))
	}
	sort.Strings(unbound)
	v.logf("the seat %s's verb %s answered with no binding on any entry (a bus module from before the field): %d bindings composed from its fields, left unbound: %s",
		BrokerSeat, BrokerListVerb, len(composed), strings.Join(unbound, ", "))
	return out, composed, unbound
}

// composeBinding is §9's binding of one entry: only for the storage types whose binding the answer names whole, and only
// when the entry's settings hold the bus object its bus name names (KV_ and the bus name, OBJ_ and the bus name).
func composeBinding(e meshbroker.Held) (*meshbroker.Binding, bool) {
	b := &meshbroker.Binding{Local: e.LocalName, Storage: e.StorageType, Template: e.Template}
	switch e.StorageType {
	case "key-value":
		if _, held := e.Settings["KV_"+e.BusName]; !held {
			return nil, false
		}
		b.Bucket = e.BusName
	case "object-store":
		if _, held := e.Settings["OBJ_"+e.BusName]; !held {
			return nil, false
		}
		b.ObjectStore = e.BusName
	default:
		return nil, false
	}
	return b, e.LocalName != "" && e.BusName != ""
}

// take writes the bindings of the entries an event did not name since the ask was sent, keeps every binding the answer does
// not list and raises its unlisted condition, and answers how many bindings it wrote and rewrote; composed names the
// entries whose binding compose gave them. The caller holds the part's bindMu.
func (v *BindingsView) take(p *Part, entries []meshbroker.Held, composed map[string]bool) (written, rewritten, fromFields int, err error) {
	v.mu.Lock()
	touched := copyOf(v.touched)
	v.mu.Unlock()
	listed := map[string]bool{}
	byModule := map[string][]meshbroker.Held{}
	for _, e := range entries {
		listed[localKey(e.Module, e.LocalName)] = true
		if e.Binding == nil {
			continue
		}
		byModule[e.Module] = append(byModule[e.Module], e)
	}
	modules := make([]string, 0, len(byModule))
	for m := range byModule {
		modules = append(modules, m)
	}
	sort.Strings(modules)
	for _, module := range modules {
		if strings.Contains(module, ".") || module == "" {
			continue
		}
		b, berr := p.bound(module)
		if berr != nil {
			return written, rewritten, fromFields, berr
		}
		changed := false
		for _, e := range byModule[module] {
			if touched[localKey(module, e.LocalName)] || e.LocalName == "" {
				continue
			}
			bind, ok := membershipBinding(e)
			if !ok {
				v.logf("the binding of %s %s in the answer of %s does not read: not written", module, e.LocalName, CallBrokerList)
				continue
			}
			cause := "the first view of the bindings (" + CallBrokerList + ")"
			if composed[localKey(module, e.LocalName)] {
				cause = "the first view of the bindings, composed from " + CallBrokerList + "'s fields"
			}
			was, had := b.Bindings[e.LocalName]
			if had && sameBinding(was, bind) {
				continue
			}
			b.Bindings[e.LocalName] = bind
			changed = true
			if composed[localKey(module, e.LocalName)] {
				fromFields++
			}
			if had {
				rewritten++
				v.logf("the binding of %s %s was rewritten, the cause %s", module, e.LocalName, cause)
			} else {
				written++
				v.logf("the binding of %s %s was written, the cause %s", module, e.LocalName, cause)
			}
		}
		if changed {
			raw, _ := json.Marshal(b)
			if _, perr := p.Store.Put(StateMemberships, BoundKey(module), raw); perr != nil {
				return written, rewritten, fromFields, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, BoundKey(module), perr)
			}
		}
	}
	return written, rewritten, fromFields, v.unlisted(p, listed, touched)
}

// unlisted keeps every binding the answer does not list and raises its condition, one per binding; a binding it lists again
// clears it (§4).
func (v *BindingsView) unlisted(p *Part, listed, touched map[string]bool) error {
	keys, err := p.Store.Keys(StateMemberships)
	if err != nil {
		return fmt.Errorf("its own %s cannot be listed: %w", StateMemberships, err)
	}
	holds := map[string]bool{}
	for _, key := range keys {
		module, ok := strings.CutPrefix(key, BoundPrefix)
		if !ok {
			continue
		}
		b, berr := p.bound(module)
		if berr != nil {
			return berr
		}
		for local := range b.Bindings {
			holds[localKey(module, local)] = true
		}
	}
	v.mu.Lock()
	if v.raised == nil {
		v.raised = map[string]bool{}
	}
	var raise, clearing []string
	for k := range holds {
		switch {
		case listed[k] && v.raised[k]:
			clearing = append(clearing, k)
		case !listed[k] && !touched[k] && !v.raised[k]:
			raise = append(raise, k)
		}
	}
	for k := range v.raised { // a raised one the part no longer holds is gone with its binding
		if !holds[k] {
			clearing = append(clearing, k)
		}
	}
	for _, k := range raise {
		v.raised[k] = true
	}
	for _, k := range clearing {
		delete(v.raised, k)
	}
	v.mu.Unlock()
	sort.Strings(raise)
	sort.Strings(clearing)
	for _, k := range raise {
		module, local, _ := strings.Cut(k, "/")
		v.logf("the binding of %s %s is held and the bus module no longer lists it: kept, and %s raised", module, local, BindingUnlisted)
		v.say(unlistedCondition(module, local, true))
	}
	for _, k := range clearing {
		module, local, _ := strings.Cut(k, "/")
		v.say(unlistedCondition(module, local, false))
	}
	return nil
}

func copyOf(m map[string]bool) map[string]bool {
	out := make(map[string]bool, len(m))
	for k, v := range m {
		out[k] = v
	}
	return out
}

// membershipBinding is an entry's binding in the form the membership holds: the seat's Binding is that form, so the JSON is
// the same. Its local name and storage are the entry's where it left them out.
func membershipBinding(e meshbroker.Held) (membership.BusBinding, bool) {
	var b membership.BusBinding
	raw, err := json.Marshal(e.Binding)
	if err != nil || json.Unmarshal(raw, &b) != nil {
		return b, false
	}
	b.Local = e.LocalName
	if b.Storage == "" {
		b.Storage = e.StorageType
	}
	if b.Template == "" {
		b.Template = e.Template
	}
	return b, true
}

// bindingsUnansweredCondition is FirstViewUnanswered for the first view of the bindings, naming the seat mesh-broker and its
// verb list.
func bindingsUnansweredCondition(raised bool, s BindingsStatus) oc.Condition {
	c := oc.Condition{Key: FirstViewUnanswered + "." + BrokerSeat + "." + FirstViewBindings, Subject: oc.ConditionSubject{Scope: "module", ID: Module}}
	if !raised {
		c.State = occ.StateCleared
		c.Resolved = CallBrokerList + " answered"
		return c
	}
	c.State, c.Severity = occ.StateRaised, occ.SeverityWarning
	c.Summary = fmt.Sprintf("the grants part has no first view of the bindings: the seat %s's verb %s has not answered since %s (%s)",
		BrokerSeat, BrokerListVerb, s.Asked.Format(time.RFC3339), strings.TrimSpace(s.Error))
	c.Headline = "The grants part waits for the bus module"
	needsYou(&c, "a look at the module holding the seat "+BrokerSeat+": is it running and connected?", "The grants part learns a binding it never "+
		"heard from the seat's "+BrokerListVerb+", and keeps asking it; it composes every account and keeps every binding it holds meanwhile, and "+
		"takes the bus module's events as they come. The answer clears this.")
	return c
}

// unlistedCondition is BindingUnlisted for one binding the part holds that the bus module no longer lists, raised or cleared.
func unlistedCondition(module, local string, raised bool) oc.Condition {
	c := oc.Condition{Key: BindingUnlisted + "." + module + "." + local, Subject: oc.ConditionSubject{Scope: "module", ID: Module}}
	if !raised {
		c.State = occ.StateCleared
		c.Resolved = "the bus module lists " + module + " " + local + " again, or dropped it"
		return c
	}
	c.State, c.Severity = occ.StateRaised, occ.SeverityWarning
	c.Summary = fmt.Sprintf("the grants part holds the binding of %s %s, which the seat %s's verb %s no longer lists", module, local, BrokerSeat, BrokerListVerb)
	c.Headline = "A binding the bus module no longer lists"
	needsYou(&c, "read the entry with the seat "+BrokerSeat+"'s show; an entry its manifest no longer asks for leaves by the bus module's drop", "The grants "+
		"part keeps the binding: it removes one only on the bus module's dropped, so a bus module whose own state was lost or rebuilt strips "+
		"no module of its data. A later answer that lists the entry, or its dropped, clears this.")
	return c
}