mesh-catalog/modules/nats/internal/keeper/keeper.go

package keeper

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

	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/membership"
)

// Module is the bus module's own name: the publisher of its events and its conditions.
const Module = "nats"

// The keeper's own states (old `state` sections, ADR 0595 §10): what it last heard per module, and the persons'
// requests its tools hand it.
const (
	StateAsked    = "asked"
	StateRequests = "requests"
	// KeyStatus is the keeper's own status in StateAsked, read by its health tool: a key no module can have, since a
	// module's name holds no underscore.
	KeyStatus = "_keeper"
)

// The bus module's events (ADR 0595 §6), on its own stream (ADR 0620 §2).
const (
	EventCreated      = "created"
	EventRefused      = "refused"
	EventDrifted      = "drifted"
	EventDriftCleared = "drift-cleared"
	EventFull         = "full"
	EventFullCleared  = "full-cleared"
	EventOrphaned     = "orphaned"
	EventRecreated    = "recreated"
	EventDropped      = "dropped"
)

// Event is the body of every bus module event, in the field names of the seat mesh-broker's definition (its message
// `said`, mesh-contracts go/seats/meshbroker Said): the entry (module, local name, storage type, template, options,
// configuration), its bus name, the settings asked and held by bus object, its fill, the snapshot a recreate or drop
// took, and why (refused, drifted). Beside the definition's fields: binding, the names a module receives (a ledger's
// history, state, indexes and snapshot, which one bus name cannot carry), and revision, the indexer's revision of the
// verdict it acted on; both asked of the definition.
//
// Written here until this module takes mesh-contracts' generated type; the swap changes no byte of the body.
type Event struct {
	V             int                        `json:"v"`
	Module        string                     `json:"module"`
	Local         string                     `json:"local-name"`
	Storage       string                     `json:"storage-type,omitempty"`
	Template      string                     `json:"template,omitempty"`
	Options       map[string]json.RawMessage `json:"options,omitempty"`
	Configuration map[string]json.RawMessage `json:"configuration,omitempty"`
	BusName       string                     `json:"bus-name"`
	Binding       membership.BusBinding      `json:"binding"`
	Asked         map[string]Settings        `json:"asked,omitempty"`
	Held          map[string]Settings        `json:"held,omitempty"`
	Fill          *EventFill                 `json:"fill,omitempty"`
	Snapshot      string                     `json:"snapshot,omitempty"`
	Why           string                     `json:"why,omitempty"`
	Revision      uint64                     `json:"revision"`
}

// EventFill is an entry's fill as an event says it: its messages, its bytes and its cap.
type EventFill struct {
	Count uint64 `json:"count"`
	Bytes uint64 `json:"bytes"`
	Cap   int64  `json:"cap"`
}

// Accepted is a manifest-accepted the keeper heard, or a module of its first view: the module, its manifest's digest,
// the indexer's revision of the verdict (0 when the event does not say it yet), and its bus-state entries whole.
type Accepted struct {
	Module   string                    `json:"module"`
	Digest   string                    `json:"digest"`
	Revision uint64                    `json:"revision"`
	BusState []catalogue.BusStateEntry `json:"bus-state"`
}

// FirstView is the indexer's answer of indexer_bus_state: every accepted manifest's bus-state entries, by module.
type FirstView struct {
	Revision uint64     `json:"revision"`
	Modules  []Accepted `json:"modules"`
}

// Asked is what the keeper holds of one module, the value of its key in StateAsked: what its newest accepted manifest
// asks for, and every entry the keeper made for it and the bus holds, by local name.
type Asked struct {
	V        int                       `json:"v"`
	Module   string                    `json:"module"`
	Digest   string                    `json:"digest,omitempty"`
	Revision uint64                    `json:"revision"`
	Entries  []catalogue.BusStateEntry `json:"entries"`
	Made     map[string]Made           `json:"made,omitempty"`
	At       time.Time                 `json:"at"`
}

// Made is one entry as the keeper made it: its plan and what stands of it now.
type Made struct {
	Plan Plan `json:"plan"`
	// Created says every object of it is on the bus and its `created` was said.
	Created bool `json:"created"`
	// Refused is why it could not be made, Waiting why it waits to be made again (a refusal the bus may lift, such as a
	// grant not composed yet).
	Refused string `json:"refused,omitempty"`
	Waiting string `json:"waiting,omitempty"`
	// Drift is how what the bus holds differs from what was asked, in a way that does not grow.
	Drift string `json:"drift,omitempty"`
	// Full is whether its fill is past the raise mark, Fill the last fill read, in percent of its cap.
	Full bool  `json:"full,omitempty"`
	Fill int64 `json:"fill,omitempty"`
	// Orphaned says no manifest asks for it any more: it goes only by drop.
	Orphaned bool `json:"orphaned,omitempty"`
}

// Status is the keeper's own state, KeyStatus: what its health tool reads.
type Status struct {
	Started time.Time `json:"started"`
	// Pass is when the last fill pass ended.
	Pass time.Time `json:"pass"`
	// FirstView is the indexer's first view: when asked, when answered and at which revision, or why not.
	FirstView FirstViewStatus `json:"first-view"`
	// Heard is when the last manifest-accepted was heard, and Unanswered since when one has waited unhandled.
	Heard      time.Time `json:"heard,omitempty"`
	Unanswered time.Time `json:"unanswered,omitempty"`
}

// FirstViewStatus is what the keeper holds of its first view.
type FirstViewStatus struct {
	Owner    string    `json:"owner"`
	Verb     string    `json:"verb"`
	Asked    time.Time `json:"asked,omitempty"`
	Answered time.Time `json:"answered,omitempty"`
	Revision uint64    `json:"revision"`
	Error    string    `json:"error,omitempty"`
	Next     time.Time `json:"next,omitempty"`
}

// The first view's owner and verb (design 25 § 12.2).
const (
	FirstViewOwner = "mesh-delivery-indexer"
	FirstViewVerb  = "indexer_bus_state"
)

// Bus is the bus as the keeper's own account reaches it.
type Bus interface {
	// Held reads the object's stream: its settings, whether it is there, and its fill (bytes and messages).
	Held(ctx context.Context, o Object) (Settings, Fill, bool, error)
	Create(ctx context.Context, o Object) error
	// Update brings a held object to o's settings: only a difference that grows (Grows), or a recreate's rebuild.
	Update(ctx context.Context, o Object) error
	Delete(ctx context.Context, o Object) error
	// Snapshot takes a snapshot of the streams and says where it is kept.
	Snapshot(ctx context.Context, streams []string) (string, error)
}

// Fill is how much one object holds: its bytes and messages.
type Fill struct {
	Bytes    uint64 `json:"bytes"`
	Messages uint64 `json:"messages"`
}

// ErrRetry marks a refusal the bus may lift without a change of the manifest (a create the keeper's account is not yet
// granted): the entry waits and is made again on the backoff, never said refused.
var ErrRetry = errors.New("refused for now")

// Store is the keeper's own state.
type Store interface {
	Get(state, key string) ([]byte, uint64, bool, error)
	Put(state, key string, value []byte) (uint64, error)
	Delete(state, key string) error
	Keys(state string) ([]string, error)
}

// Keeper is the bus module's keeper.
type Keeper struct {
	Bus   Bus
	Store Store
	// Emit says one of the bus module's events; Say raises or clears a condition through the condition keeper.
	Emit func(event string, body Event) error
	Say  func(asks.Condition) error
	// View asks the indexer's indexer_bus_state.
	View func() (FirstView, error)
	// Act records a hand act through the seat mesh-hand-acts; Warrant reads the router's record of an ask.
	Act     func(what, why, cause string) error
	Warrant func(askID string) (*asks.Ask, *asks.Warrant, error)
	Now     func() time.Time
	Logf    func(string, ...any)
	// Raise and Clear are the fill marks in percent (ADR 0595 §6): settings of the bus module, 75 and 70 by default.
	Raise, Clear int64

	mu     sync.Mutex
	status Status
	viewed bool
}

func (k *Keeper) now() time.Time {
	if k.Now == nil {
		return time.Now().UTC()
	}
	return k.Now().UTC()
}

func (k *Keeper) logf(format string, a ...any) {
	if k.Logf != nil {
		k.Logf(format, a...)
	}
}

func (k *Keeper) marks() (int64, int64) {
	raise, clear := k.Raise, k.Clear
	if raise <= 0 {
		raise = 75
	}
	if clear <= 0 || clear >= raise {
		clear = raise - 5
	}
	return raise, clear
}

// Viewed is whether the first view was answered: before it the keeper creates and orphans nothing.
func (k *Keeper) Viewed() bool {
	k.mu.Lock()
	defer k.mu.Unlock()
	return k.viewed
}

// Status is the keeper's status now.
func (k *Keeper) Status() Status {
	k.mu.Lock()
	defer k.mu.Unlock()
	return k.status
}

func (k *Keeper) setStatus(f func(*Status)) {
	k.mu.Lock()
	if k.status.Started.IsZero() {
		k.status.Started = k.now()
	}
	k.status.FirstView.Owner, k.status.FirstView.Verb = FirstViewOwner, FirstViewVerb
	f(&k.status)
	s := k.status
	k.mu.Unlock()
	if k.Store != nil {
		raw, _ := json.Marshal(s)
		if _, err := k.Store.Put(StateAsked, KeyStatus, raw); err != nil {
			k.logf("the keeper's status cannot be written to its own %s: %v", StateAsked, err)
		}
	}
}

// AskFirstView asks the indexer's first view once and applies it (design 25 § 12.2): every module it answers is
// applied as an accepted verdict, a module the keeper holds and the view leaves out is kept (a set-aside manifest is
// missing from it, never removed), and then check runs over what is held. It answers the error and when to ask again,
// on the backoff from 10 s doubling to 5 min; the keeper's health is unhealthy meanwhile, naming the verb.
func (k *Keeper) AskFirstView(wait time.Duration) (time.Duration, error) {
	at := k.now()
	k.setStatus(func(s *Status) { s.FirstView.Asked = at })
	view, err := k.View()
	if err != nil {
		next := wait * 2
		if wait <= 0 {
			next = 10 * time.Second
		}
		if next > 5*time.Minute {
			next = 5 * time.Minute
		}
		k.setStatus(func(s *Status) {
			s.FirstView.Error = fmt.Sprintf("%s.%s did not answer: %v", FirstViewOwner, FirstViewVerb, err)
			s.FirstView.Next = at.Add(next)
		})
		return next, err
	}
	k.mu.Lock()
	k.viewed = true
	k.mu.Unlock()
	k.setStatus(func(s *Status) {
		s.FirstView.Answered, s.FirstView.Revision, s.FirstView.Error, s.FirstView.Next = k.now(), view.Revision, "", time.Time{}
	})
	for _, a := range view.Modules {
		k.Heard(a)
	}
	k.CheckAll()
	return 0, nil
}

// Heard applies one accepted verdict (manifest-accepted, or a module of the first view): applied only when newer than
// what the keeper holds for that module by the indexer's revision alone, so a replay never takes an entry back (a
// verdict that says no revision is applied when its digest differs). Each entry asked is made where the bus lacks it;
// each entry made that is no longer asked is an orphan. An empty verdict ("this module asks nothing now") ends the
// module's key once nothing it made is held.
func (k *Keeper) Heard(a Accepted) {
	k.setStatus(func(s *Status) { s.Heard = k.now() })
	if !k.Viewed() {
		k.logf("manifest-accepted of %s heard before the first view: kept on the consumer until it is answered", a.Module)
		return
	}
	held, _, found := k.asked(a.Module)
	if found && a.Revision != 0 && a.Revision <= held.Revision {
		k.logf("manifest-accepted of %s at revision %d is not newer than the %d held: not applied", a.Module, a.Revision, held.Revision)
		return
	}
	if found && a.Revision == 0 && a.Digest != "" && a.Digest == held.Digest {
		return
	}
	if !found {
		held = Asked{V: 1, Module: a.Module}
	}
	if held.Made == nil {
		held.Made = map[string]Made{}
	}
	held.Digest, held.Entries, held.At = a.Digest, a.BusState, k.now()
	if a.Revision > held.Revision {
		held.Revision = a.Revision
	}
	asked := map[string]bool{}
	for _, e := range a.BusState {
		asked[e.Name] = true
		k.makeEntry(&held, e)
	}
	for _, local := range sortedKeys(held.Made) {
		m := held.Made[local]
		if asked[local] || m.Orphaned {
			continue
		}
		if !m.Created {
			// Never made: nothing on the bus to orphan.
			delete(held.Made, local)
			k.clear(conditionRefused, a.Module, local)
			continue
		}
		m.Orphaned = true
		held.Made[local] = m
		k.emit(EventOrphaned, m.Plan, held.Revision, "no accepted manifest of "+a.Module+" asks for it now", nil, nil)
		k.raise(conditionOrphaned, a.Module, local, asks.SeverityWarning, "Kept data no manifest asks for",
			"the operator's drop of it, or a manifest asking for it again",
			fmt.Sprintf("%s of %s is on the bus and no accepted manifest asks for it: it goes only by the bus module's drop, on your warrant (nats_drop), or a manifest asking for it again keeps it", BusName(a.Module, local), a.Module))
	}
	k.write(held)
}

// makeEntry makes one asked entry where the bus lacks it, compares it where it holds it, and says what it did.
func (k *Keeper) makeEntry(held *Asked, e catalogue.BusStateEntry) {
	module := held.Module
	plan, err := PlanOf(module, e)
	m := held.Made[e.Name]
	if err != nil {
		if m.Refused != err.Error() {
			k.emit(EventRefused, Plan{Module: module, Local: e.Name, Storage: e.Storage, Template: e.Template}, held.Revision, err.Error(), nil, nil)
			k.raise(conditionRefused, module, e.Name, asks.SeverityWarning, "The bus module refused to create kept data",
				"a manifest that asks for it in a form the bus module makes",
				fmt.Sprintf("%s of %s was not created: %v. The module gets no binding for it and fails at its start, naming it", e.Name, module, err))
		}
		m.Refused, m.Waiting = err.Error(), ""
		held.Made[e.Name] = m
		return
	}
	wasOrphan := m.Orphaned
	m.Plan, m.Orphaned = plan, false
	if wasOrphan {
		// Asked again: kept, and its binding said again with created (design 25 § 12.3).
		m.Created = false
		k.clear(conditionOrphaned, module, e.Name)
	}
	k.ensure(held, &m)
	held.Made[e.Name] = m
}

// ensure brings one planned entry onto the bus: each object the bus lacks created, a held one compared. A difference
// that only grows is applied; any other is a drift, said and never fixed quietly.
func (k *Keeper) ensure(held *Asked, m *Made) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	asked, holds := map[string]Settings{}, map[string]Settings{}
	var drifts []string
	for _, o := range m.Plan.Objects {
		asked[o.Name] = o.Settings
		s, _, found, err := k.Bus.Held(ctx, o)
		if err != nil {
			m.Waiting = fmt.Sprintf("%s cannot be read: %v", o.Stream(), err)
			return
		}
		if !found {
			if err := k.Bus.Create(ctx, o); err != nil {
				if errors.Is(err, ErrRetry) {
					m.Waiting = fmt.Sprintf("%s cannot be created yet: %v", o.Stream(), err)
					return
				}
				m.Refused, m.Waiting = fmt.Sprintf("%s: %v", o.Stream(), err), ""
				k.emit(EventRefused, m.Plan, held.Revision, m.Refused, asked, nil)
				k.raise(conditionRefused, held.Module, m.Plan.Local, asks.SeverityWarning, "The bus module refused to create kept data",
					"a look at why the bus refused it",
					fmt.Sprintf("%s of %s was not created: %s. The module gets no binding for it and fails at its start, naming it", m.Plan.Local, held.Module, m.Refused))
				return
			}
			holds[o.Name] = o.Settings
			continue
		}
		holds[o.Name] = s
		if Differs(o.Settings, s) == "" {
			continue
		}
		if Grows(o.Settings, s) {
			if err := k.Bus.Update(ctx, o); err != nil {
				m.Waiting = fmt.Sprintf("%s's larger settings cannot be applied yet: %v", o.Stream(), err)
				return
			}
			holds[o.Name] = o.Settings
			continue
		}
		drifts = append(drifts, o.Stream()+": "+Differs(o.Settings, s))
	}
	wasRefused := m.Refused != ""
	m.Waiting, m.Refused = "", ""
	if wasRefused {
		k.clear(conditionRefused, held.Module, m.Plan.Local)
	}
	if !m.Created {
		m.Created = true
		k.emit(EventCreated, m.Plan, held.Revision, "", asked, holds)
	}
	drift := strings.Join(drifts, "; ")
	switch {
	case drift != "" && drift != m.Drift:
		m.Drift = drift
		k.emit(EventDrifted, m.Plan, held.Revision, drift, asked, holds)
		k.raise(conditionDrifted, held.Module, m.Plan.Local, asks.SeverityWarning, "Kept data differs from what was asked",
			"a decision: change the manifest back, or the bus module's recreate on your warrant",
			fmt.Sprintf("%s of %s differs from what its manifest asks, in a way that does not only grow: %s", m.Plan.Local, held.Module, drift))
	case drift == "" && m.Drift != "":
		m.Drift = ""
		k.emit(EventDriftCleared, m.Plan, held.Revision, "", asked, holds)
		k.clear(conditionDrifted, held.Module, m.Plan.Local)
	}
}

// CheckAll compares every entry the keeper holds with the bus, making what is missing (design 25 § 12.2's check at
// start) and making again what waits; it answers what it found, one line per entry.
func (k *Keeper) CheckAll() []CheckLine {
	var out []CheckLine
	keys, err := k.Store.Keys(StateAsked)
	if err != nil {
		return []CheckLine{{Why: "the keeper's own " + StateAsked + " cannot be read: " + err.Error()}}
	}
	for _, module := range keys {
		if module == KeyStatus {
			continue
		}
		held, _, found := k.asked(module)
		if !found {
			continue
		}
		for _, local := range sortedKeys(held.Made) {
			m := held.Made[local]
			if m.Orphaned || m.Plan.Local == "" {
				out = append(out, lineOf(module, local, m))
				continue
			}
			for _, e := range held.Entries {
				if e.Name == local {
					k.makeEntry(&held, e)
				}
			}
			out = append(out, lineOf(module, local, held.Made[local]))
		}
		k.write(held)
	}
	return out
}

// CheckLine is one entry as check answers it.
type CheckLine struct {
	Module   string `json:"module,omitempty"`
	Local    string `json:"local,omitempty"`
	Bus      string `json:"bus,omitempty"`
	Created  bool   `json:"created"`
	Refused  string `json:"refused,omitempty"`
	Waiting  string `json:"waiting,omitempty"`
	Drift    string `json:"drift,omitempty"`
	Orphaned bool   `json:"orphaned,omitempty"`
	Full     bool   `json:"full,omitempty"`
	Fill     int64  `json:"fill-percent,omitempty"`
	Why      string `json:"why,omitempty"`
}

func lineOf(module, local string, m Made) CheckLine {
	return CheckLine{Module: module, Local: local, Bus: BusName(module, local), Created: m.Created, Refused: m.Refused,
		Waiting: m.Waiting, Drift: m.Drift, Orphaned: m.Orphaned, Full: m.Full, Fill: m.Fill}
}

// Waiting is every entry that waits to be made again, `<module> <local>: <why>`: what the backoff retries.
func (k *Keeper) Waiting() []string {
	var out []string
	keys, _ := k.Store.Keys(StateAsked)
	for _, module := range keys {
		held, _, found := k.asked(module)
		if !found {
			continue
		}
		for _, local := range sortedKeys(held.Made) {
			if w := held.Made[local].Waiting; w != "" {
				out = append(out, module+" "+local+": "+w)
			}
		}
	}
	return out
}

// The kinds of condition the keeper raises, each naming the module and the entry (design 25 § 12.3).
const (
	conditionFull     = "full"
	conditionDrifted  = "drifted"
	conditionOrphaned = "orphaned"
	conditionRefused  = "refused"
)

// ConditionKey is the key of one of the keeper's conditions: `nats.<kind>.<module>_<local>`.
func ConditionKey(kind, module, local string) string {
	return Module + "." + kind + "." + BusName(module, local)
}

func (k *Keeper) raise(kind, module, local, severity, headline, needs, explanation string) {
	if k.Say == nil {
		return
	}
	c := asks.Condition{Key: ConditionKey(kind, module, local), Subject: asks.Scope{Scope: "module", ID: module},
		State: asks.StateRaised, Severity: severity, Headline: headline, Needs: needs,
		Summary:     explanation,
		Explanation: asks.NeedsYou + needs + ". " + explanation,
		Resolved:    "cleared by the bus module once it no longer holds"}
	if err := k.Say(c); err != nil {
		k.logf("the condition %s cannot be raised: %v", c.Key, err)
	}
}

func (k *Keeper) clear(kind, module, local string) {
	if k.Say == nil {
		return
	}
	c := asks.Condition{Key: ConditionKey(kind, module, local), Subject: asks.Scope{Scope: "module", ID: module}, State: asks.StateCleared}
	if err := k.Say(c); err != nil {
		k.logf("the condition %s cannot be cleared: %v", c.Key, err)
	}
}

func (k *Keeper) emit(event string, p Plan, revision uint64, why string, asked, held map[string]Settings) {
	if k.Emit == nil {
		return
	}
	body := Event{V: 1, Module: p.Module, Local: p.Local, Storage: p.Storage, Template: p.Template, Options: p.Entry.Options,
		Configuration: p.Entry.Configuration, BusName: BusName(p.Module, p.Local), Binding: p.Binding,
		Asked: asked, Held: held, Why: why, Revision: revision}
	if event == EventDropped {
		// A dropped entry's names only: the binding to remove.
		body.Binding.Types = nil
	}
	if err := k.Emit(event, body); err != nil {
		k.logf("the event %s of %s %s could not be said: %v", event, p.Module, p.Local, err)
	}
}

func (k *Keeper) asked(module string) (Asked, uint64, bool) {
	raw, rev, found, err := k.Store.Get(StateAsked, module)
	if err != nil || !found {
		return Asked{}, 0, false
	}
	var a Asked
	if json.Unmarshal(raw, &a) != nil {
		return Asked{}, 0, false
	}
	return a, rev, true
}

// write keeps a module's key, or removes it when its manifest asks for nothing and nothing it made is held: a key goes
// only so, never because a module is missing from a first view.
func (k *Keeper) write(a Asked) {
	if len(a.Entries) == 0 && len(a.Made) == 0 {
		if err := k.Store.Delete(StateAsked, a.Module); err != nil {
			k.logf("the key of %s cannot be removed from its own %s: %v", a.Module, StateAsked, err)
		}
		return
	}
	raw, _ := json.Marshal(a)
	if _, err := k.Store.Put(StateAsked, a.Module, raw); err != nil {
		k.logf("the key of %s cannot be written to its own %s: %v", a.Module, StateAsked, err)
	}
}

// Held is every module the keeper holds, by name: what list, show and orphans answer from.
func (k *Keeper) Held() map[string]Asked {
	out := map[string]Asked{}
	keys, _ := k.Store.Keys(StateAsked)
	for _, module := range keys {
		if module == KeyStatus {
			continue
		}
		if a, _, found := k.asked(module); found {
			out[module] = a
		}
	}
	return out
}

func sortedKeys[V any](m map[string]V) []string {
	out := make([]string, 0, len(m))
	for key := range m {
		out = append(out, key)
	}
	sort.Strings(out)
	return out
}