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

package grants

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

	"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"mesh-grants/internal/buscompose"
)

// **The recovery module's restore points' names, read through `kept-points`** (design 47 G §565, A-230). The bus names of the
// recovery module's account include those a restore point (or an aside) names. The part reads them by its `invokes` of
// `seat:mesh-recovery.kept-points`, a kept call (ADR 0668 §1: made again until it answers), at its start and again on every `ended`
// with outcome `done`, every `point-removed` and every `expiry-ended` of the seat, which it consumes. Until the call first answers
// after a start, the part grants the names it last granted from it, kept in its state memberships (KeptNamesKey): it takes no name
// away before that answer, and grants_first_views says the wait.

// KeptPointsVerb is the seat's verb as the manifest's invokes names it.
const (
	CallKeptPoints = "seat:" + meshrecovery.Seat + "." + meshrecovery.VerbKeptPoints
	// KeptNamesKey is the key of the part's own state memberships that keeps the names last read from `kept-points`; never a
	// `<node>.<module>` key, since no machine is named `kept-names`.
	KeptNamesKey = "kept-names." + meshrecovery.Seat
)

// KeptNames is the value of KeptNamesKey: every bus name a restore point or an aside of the recovery module names.
type KeptNames struct {
	V     int       `json:"v"`
	Names []string  `json:"names"`
	At    time.Time `json:"at,omitzero"`
}

// keptPointsAnswer is the part of kept-points' answer the part reads: `points[].names` and `asides[].names` (A-230). The contract's
// Point gains `names` at the join, as its Aside has it; until then this is the wire's form.
type keptPointsAnswer struct {
	Points []struct {
		Names []string `json:"names"`
	} `json:"points"`
	Asides []struct {
		Names []string `json:"names"`
	} `json:"asides"`
}

// KeptPointsStatus is what grants_first_views says of the kept call.
type KeptPointsStatus struct {
	Seat     string    `json:"seat"`
	Verb     string    `json:"verb"`
	Asked    time.Time `json:"asked,omitzero"`
	Answered time.Time `json:"answered,omitzero"`
	Asks     int       `json:"asks"`
	// Names is how many names the last answer named (or, while waiting, the names kept from the last answer before the start).
	Names int `json:"names"`
	// Waiting says the call has not answered since the start: the part grants the names it kept.
	Waiting bool   `json:"waiting"`
	Why     string `json:"why,omitempty"`
	Error   string `json:"error,omitempty"`
}

// KeptPointsView is the kept call of `kept-points`.
type KeptPointsView struct {
	// Ask calls the seat's verb and answers its reply (the process's connection).
	Ask     func(ctx context.Context) (json.RawMessage, error)
	Changed func()
	Logf    func(string, ...any)
	Now     func() time.Time
	// Pause waits d between two asks that failed and answers false once ctx ended; nil waits on the context's bound.
	Pause func(ctx context.Context, d time.Duration) bool
	// Wait is the first wait between two asks that failed (ten seconds when zero), doubling to MaxWait (five minutes when zero).
	Wait    time.Duration
	MaxWait time.Duration

	mu     sync.Mutex
	ctx    context.Context
	part   *Part
	st     KeptPointsStatus
	asking bool
	again  bool
}

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

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

// Status is the kept call's status now.
func (v *KeptPointsView) Status() KeptPointsStatus {
	v.mu.Lock()
	defer v.mu.Unlock()
	s := v.st
	s.Seat, s.Verb = meshrecovery.Seat, meshrecovery.VerbKeptPoints
	return s
}

// Start asks at the part's start, on a goroutine of its own, and until it answers.
func (v *KeptPointsView) Start(ctx context.Context, p *Part) {
	v.mu.Lock()
	v.ctx, v.part = ctx, p
	v.st.Waiting = true
	v.mu.Unlock()
	if kept, err := p.keptNames(); err == nil {
		v.mu.Lock()
		v.st.Names = len(kept)
		v.mu.Unlock()
	}
	go v.askUntil(ctx, p, "start")
}

// Heard asks again when an event heard is one that changes the points: `ended` with outcome `done`, `point-removed`, `expiry-ended`
// of the seat mesh-recovery, named `mesh-recovery.<event>` or by its bus subject.
func (v *KeptPointsView) Heard(event string, body []byte) {
	name, ok := recoveryEvent(event)
	if !ok || !changesPoints(name, body) {
		return
	}
	v.mu.Lock()
	ctx, p := v.ctx, v.part
	v.mu.Unlock()
	if p == nil {
		return // not started: the start's ask reads what the event changed
	}
	go v.askUntil(ctx, p, "the seat's "+name)
}

func recoveryEvent(event string) (string, bool) {
	for _, prefix := range []string{meshrecovery.Seat + ".", "mesh.seat." + meshrecovery.Seat + ".event."} {
		if name, ok := strings.CutPrefix(event, prefix); ok {
			return name, true
		}
	}
	return "", false
}

func changesPoints(name string, body []byte) bool {
	switch name {
	case meshrecovery.EventPointRemoved, meshrecovery.EventExpiryEnded:
		return true
	case meshrecovery.EventEnded:
		var e meshrecovery.EndedEvent
		return json.Unmarshal(body, &e) != nil || e.Outcome == meshrecovery.EndedEventOutcomeDone
	}
	return false
}

// IsKeptPointsEvent says an event heard is one the kept call asks again on.
func IsKeptPointsEvent(event string, body []byte) bool {
	name, ok := recoveryEvent(event)
	return ok && changesPoints(name, body)
}

func (v *KeptPointsView) 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 seat answers or ctx ends, from ten seconds, doubling, at most every five minutes. An ask made while one
// runs is made once more when it ends, so an event heard meanwhile is never lost.
func (v *KeptPointsView) askUntil(ctx context.Context, p *Part, why string) {
	v.mu.Lock()
	if v.asking {
		v.again = true
		v.mu.Unlock()
		return
	}
	v.asking = true
	v.st.Why = why
	v.mu.Unlock()
	wait, most := v.Wait, v.MaxWait
	if wait == 0 {
		wait = 10 * time.Second
	}
	if most == 0 {
		most = 5 * time.Minute
	}
	for {
		v.mu.Lock()
		v.st.Asked = v.now()
		v.st.Asks++
		v.mu.Unlock()
		raw, err := v.Ask(ctx)
		var n int
		if err == nil {
			n, err = p.applyKeptPoints(raw, v.wake)
		}
		if err != nil {
			v.mu.Lock()
			v.st.Error = err.Error()
			v.mu.Unlock()
			v.logf("%s is not answered (%v): asked again in %s; the names last kept stay granted", CallKeptPoints, err, wait)
			if !v.pause(ctx, wait) {
				v.mu.Lock()
				v.asking = false
				v.mu.Unlock()
				return
			}
			if wait *= 2; wait > most {
				wait = most
			}
			continue
		}
		v.mu.Lock()
		v.st.Answered, v.st.Waiting, v.st.Error, v.st.Names = v.now(), false, "", n
		again := v.again
		v.again = false
		if !again {
			v.asking = false
		}
		v.mu.Unlock()
		v.wake()
		if !again {
			return
		}
		wait = v.Wait
		if wait == 0 {
			wait = 10 * time.Second
		}
	}
}

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

// keptNames are the names last read from kept-points, from the part's own state: none before the first answer ever.
func (p *Part) keptNames() ([]string, error) {
	raw, _, found, err := p.Store.Get(StateMemberships, KeptNamesKey)
	if err != nil {
		return nil, fmt.Errorf("its own %s %s cannot be read: %w", StateMemberships, KeptNamesKey, err)
	}
	if !found {
		return nil, nil
	}
	var k KeptNames
	if err := json.Unmarshal(raw, &k); err != nil {
		return nil, fmt.Errorf("its own %s %s does not read: %w", StateMemberships, KeptNamesKey, err)
	}
	return k.Names, nil
}

// applyKeptPoints replaces the kept names with those of an answer (a name no point names any more is taken away), and calls wake
// when they changed. It answers how many names the answer named.
func (p *Part) applyKeptPoints(raw json.RawMessage, wake func()) (int, error) {
	var a keptPointsAnswer
	if err := json.Unmarshal(raw, &a); err != nil {
		return 0, fmt.Errorf("%s answered what it does not read: %w", CallKeptPoints, err)
	}
	var names []string
	for _, pt := range a.Points {
		names = append(names, pt.Names...)
	}
	for _, as := range a.Asides {
		names = append(names, as.Names...)
	}
	slices.Sort(names)
	names = slices.Compact(names)
	was, err := p.keptNames()
	if err != nil {
		return 0, err
	}
	if slices.Equal(was, names) {
		return len(names), nil
	}
	body, _ := json.Marshal(KeptNames{V: 1, Names: names, At: p.now()})
	if _, err := p.Store.Put(StateMemberships, KeptNamesKey, body); err != nil {
		return 0, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, KeptNamesKey, err)
	}
	if wake != nil {
		wake()
	}
	return len(names), nil
}

// withKeptNames sets the names the recovery module's account is granted for restore points on in (Inputs.KeptNames), read from
// the part's own state in one place for the pass and for its tools.
func (p *Part) withKeptNames(in *Inputs) error {
	names, err := p.keptNames()
	if err != nil {
		return err
	}
	in.KeptNames = names
	return nil
}

// recoveryNamesOf are the bus names of kept data of every manifest of the catalog graph the part holds, and the names a restore
// point or an aside names (Inputs.KeptNames): a name a manifest left stays while a point names it. Sorted, once each. Pure.
func recoveryNamesOf(in Inputs) []string {
	names := slices.Clone(in.KeptNames)
	for _, group := range []map[string]catalogue.Manifest{in.Manifests, in.Running, in.Ran, in.RanByFingerprint, in.Had, in.Held} {
		for _, m := range group {
			names = append(names, buscompose.KeptBusNames(m)...)
		}
	}
	slices.Sort(names)
	return slices.Compact(names)
}