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

package grants

import (
	"encoding/json"
	"errors"
	"fmt"

	"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
	stdio "git.novox.be/novox/mesh-sdk/go"
	"slices"
	"sort"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/membership"
	"mesh-grants/internal/buscompose"
)

// **The binding travels on the bus module's events** (novox/hq ADR 0595 §1, §7; design 25 § 12.2; DLV-G1). The bus
// module (the holder of the seat mesh-broker) says `created` when it created one entry of a module's bus state,
// `recreated` when it rebuilt one, and `dropped` when it removed one; each carries the entry's names on the bus. The
// grants part hears them as the seat's events, `mesh-broker.<event>` on the shared events stream (its manifest's
// `consumes`), on its durable consumer; writes the binding into the module's membership on every machine the module is
// assigned, rewrites it on `recreated` and removes it on `dropped`. What it heard is kept in its own state memberships,
// one key per module (BoundKey), so a binding outlives the 7 days the shared events stream keeps its event and reaches a
// machine the module is assigned to later. A `refused` create hands no binding: only these three events are read.

// The bus module's events the part hears, by their local names on the seat mesh-broker (ADR 0595 §6).
const (
	BrokerCreated   = "created"
	BrokerRecreated = "recreated"
	BrokerDropped   = "dropped"
)

// BoundPrefix starts the key of the part's own state memberships that keeps one module's bindings, `bound.<module>`:
// never a `<node>.<module>` key, since no machine is named `bound` (a membership's runtime reads its own key alone).
const BoundPrefix = "bound."

// BoundKey is the key keeping module's bindings.
func BoundKey(module string) string { return BoundPrefix + module }

// BrokerEvent is the body of the bus module's `created`, `recreated` and `dropped`, the fields the part reads.
//
// It stands until the seat mesh-broker's definition is generated in mesh-contracts (ADR 0613 §1, §5; design 26): this
// part then reads the generated type, with the same fields.
type BrokerEvent struct {
	V      int    `json:"v,omitempty"`
	Module string `json:"module"`
	// LocalName and StorageType are the seat's contract's names (mesh-contracts meshbroker.Said); Local and Storage the
	// same read where an older body names them so.
	LocalName   string `json:"local-name,omitempty"`
	Local       string `json:"local,omitempty"`
	StorageType string `json:"storage-type,omitempty"`
	Storage     string `json:"storage,omitempty"`
	Template    string `json:"template,omitempty"`
	// Binding is the entry's names on the bus: whole on `created` and `recreated`; on `dropped` only its local name counts.
	Binding membership.BusBinding `json:"binding"`
}

// Bound is the value of BoundKey: every binding of one module the bus module said it created and has not dropped, by
// local name.
type Bound struct {
	V        int                              `json:"v"`
	Module   string                           `json:"module"`
	Bindings map[string]membership.BusBinding `json:"bindings"`
}

// brokerEvent is the local name of an event heard from the holder of mesh-broker, by its bus subject, and whether it is
// one: the seat's own subject alone (`mesh.seat.mesh-broker.event.<event>`, the lead's ruling, registry milestone 13:0x),
// where only its holder publishes, and only while a module holds the seat. The bus module's own stream is its history
// for its tool, never what the part hears.
func brokerEvent(event string, brokers map[string]bool) (string, bool) {
	name, onTheSeat := strings.CutPrefix(event, "mesh.seat."+meshbroker.Seat+".event.")
	if !onTheSeat || len(brokers) == 0 {
		return "", false
	}
	switch name {
	case BrokerCreated, BrokerRecreated, BrokerDropped:
		return name, true
	}
	return "", false
}

// HearBroker applies one event heard to the kept bindings: a `created` or `recreated` writes the entry's binding, a
// `dropped` removes it. It answers whether the kept bindings changed (so a pass composes again), or why the event was
// not applied; an event that is no bus module's is no change and no error. brokers are the modules holding mesh-broker.
func (p *Part) HearBroker(event string, body []byte, brokers map[string]bool) (bool, error) {
	name, ok := brokerEvent(event, brokers)
	if !ok {
		return false, nil
	}
	var e BrokerEvent
	if err := json.Unmarshal(body, &e); err != nil {
		return false, fmt.Errorf("the bus module's %s does not read: %w", name, err)
	}
	local := firstOf(e.LocalName, e.Local, e.Binding.Local)
	storage := firstOf(e.StorageType, e.Storage)
	if e.Module == "" || local == "" || strings.Contains(e.Module, ".") {
		return false, fmt.Errorf("the bus module's %s names no module or no entry (%q, %q)", name, e.Module, local)
	}
	// Written at the revision read, and read again once when another write came first (two `created` of one module heard
	// at once): a binding is never lost to a write it did not see (issue 1966).
	changed, err := p.keepBinding(e, name, local, storage)
	if errors.Is(err, stdio.ErrChanged) {
		changed, err = p.keepBinding(e, name, local, storage)
	}
	return changed, err
}

// keepBinding applies one event to the kept bindings of its module, written at the revision it read them at.
func (p *Part) keepBinding(e BrokerEvent, name, local, storage string) (bool, error) {
	b, rev, err := p.boundAt(e.Module)
	if err != nil {
		return false, err
	}
	was, had := b.Bindings[local]
	switch name {
	case BrokerDropped:
		if !had {
			return false, nil
		}
		delete(b.Bindings, local)
	default:
		bind := e.Binding
		bind.Local = local
		if bind.Storage == "" {
			bind.Storage = storage
		}
		if bind.Template == "" {
			bind.Template = e.Template
		}
		if had && sameBinding(was, bind) {
			return false, nil
		}
		b.Bindings[local] = bind
	}
	raw, _ := json.Marshal(b)
	if _, err := p.Store.Update(StateMemberships, BoundKey(e.Module), raw, rev); err != nil {
		return false, fmt.Errorf("its own %s %s cannot be written: %w", StateMemberships, BoundKey(e.Module), err)
	}
	return true, nil
}

// bound reads the kept bindings of module; none kept is an empty value.
func (p *Part) bound(module string) (Bound, error) {
	b, _, err := p.boundAt(module)
	return b, err
}

// boundAt is bound with the revision it was read at, 0 when none is kept.
func (p *Part) boundAt(module string) (Bound, uint64, error) {
	b := Bound{V: 1, Module: module, Bindings: map[string]membership.BusBinding{}}
	raw, rev, found, err := p.Store.Get(StateMemberships, BoundKey(module))
	if err != nil {
		return b, 0, fmt.Errorf("its own %s %s cannot be read: %w", StateMemberships, BoundKey(module), err)
	}
	if found {
		if err := json.Unmarshal(raw, &b); err != nil {
			return b, 0, fmt.Errorf("its own %s %s does not read: %w", StateMemberships, BoundKey(module), err)
		}
		if b.Bindings == nil {
			b.Bindings = map[string]membership.BusBinding{}
		}
	}
	return b, rev, nil
}

// bindingsOf is module's kept bindings in local-name order, for its memberships; nil when none.
func (p *Part) bindingsOf(module string) ([]membership.BusBinding, error) {
	b, err := p.bound(module)
	if err != nil || len(b.Bindings) == 0 {
		return nil, err
	}
	locals := make([]string, 0, len(b.Bindings))
	for l := range b.Bindings {
		locals = append(locals, l)
	}
	sort.Strings(locals)
	out := make([]membership.BusBinding, 0, len(locals))
	for _, l := range locals {
		out = append(out, b.Bindings[l])
	}
	return out, nil
}

// BindingsOf answers the tool grants_bindings: every binding the part keeps of one module, or of every module when module
// is empty (its own state memberships, key BoundKey), each beside the memberships `<node>.<module>` it wrote, and by
// membership the kept local names its bindings lack, so a binding kept but not yet written into a membership is named.
func (p *Part) BindingsOf(module string) (any, error) {
	keys, err := p.Store.Keys(StateMemberships)
	if err != nil {
		return nil, fmt.Errorf("its own %s cannot be listed: %w", StateMemberships, err)
	}
	type kept struct {
		Module      string                  `json:"module"`
		Bindings    []membership.BusBinding `json:"bindings"`
		Memberships map[string][]string     `json:"memberships"`
		Lacking     map[string][]string     `json:"lacking,omitempty"`
	}
	out := []kept{}
	for _, key := range keys {
		mod, ok := strings.CutPrefix(key, BoundPrefix)
		if !ok || (module != "" && mod != module) {
			continue
		}
		binds, err := p.bindingsOf(mod)
		if err != nil {
			return nil, err
		}
		k := kept{Module: mod, Bindings: binds, Memberships: map[string][]string{}}
		if k.Bindings == nil {
			k.Bindings = []membership.BusBinding{}
		}
		for _, mk := range keys {
			if _, of, _ := strings.Cut(mk, "."); strings.HasPrefix(mk, BoundPrefix) || of != mod {
				continue
			}
			raw, _, found, err := p.Store.Get(StateMemberships, mk)
			if err != nil {
				return nil, fmt.Errorf("its own membership %s cannot be read: %w", mk, err)
			}
			if !found {
				continue
			}
			bare, err := membership.BareGrantsMembership(raw)
			if err != nil {
				return nil, fmt.Errorf("its own membership %s does not read: %w", mk, err)
			}
			var m membership.Membership
			if err := json.Unmarshal(bare, &m); err != nil {
				return nil, fmt.Errorf("its own membership %s does not read: %w", mk, err)
			}
			written := []string{}
			for _, b := range m.Bindings {
				written = append(written, b.Local)
			}
			k.Memberships[mk] = written
			for _, b := range binds {
				if !slices.Contains(written, b.Local) {
					if k.Lacking == nil {
						k.Lacking = map[string][]string{}
					}
					k.Lacking[mk] = append(k.Lacking[mk], b.Local)
				}
			}
		}
		out = append(out, k)
	}
	again, err := p.lastBindingsAgain(module)
	if err != nil {
		return nil, err
	}
	answer := map[string]any{"modules": out,
		"says": "each binding the part kept from the bus module's created and recreated, beside the memberships it wrote; lacking names a kept binding a membership does not carry yet"}
	if module != "" && len(out) == 0 {
		answer["says"] = "the part keeps no binding of " + module + ": the bus module said no created for it that the part heard"
	}
	if again != nil {
		answer["read-again"] = again
	}
	return answer, nil
}

// **A binding the part did not keep is read again from the shared events stream** (issue 1966; the lead's ruling, option
// b): grants_bindings_again asks the part's process, by a person's act recorded as a hand act, to read the seat
// mesh-broker's events the stream still holds, oldest first, and to keep each of one module's created, recreated and
// dropped through HearBroker, as if heard now, so a binding dropped since is not brought back. The stream keeps its events
// for a bounded age: a created older than that is not there, and the answer says so, naming the path then (the seat's
// list answering each entry's binding, a record extending ADR 0595 §7).

// KeyBindingsAgain is, in StateUserList, the last re-read of one module's bindings: BindingsAgain.
const KeyBindingsAgain = "bindings-again"

// BrokerHistory reads the seat mesh-broker's events again from the shared events stream.
type BrokerHistory interface {
	// Said is every event of the seat the stream still holds, oldest first.
	Said() ([]SaidEvent, error)
}

// SaidEvent is one event of the seat mesh-broker as the shared events stream holds it.
type SaidEvent struct {
	Subject string    `json:"subject"`
	Data    []byte    `json:"data"`
	At      time.Time `json:"at"`
}

// BindingsAgain is what the last re-read of one module's bindings found: the seat's events read, the local names whose
// binding it kept (Kept), those it found already kept as said (Already), those a dropped took away (Dropped), the oldest
// event the stream still holds, and in plain words what it means.
type BindingsAgain struct {
	V       int       `json:"v"`
	Module  string    `json:"module"`
	By      string    `json:"by"`
	At      time.Time `json:"at"`
	Read    int       `json:"read"`
	Kept    []string  `json:"kept"`
	Already []string  `json:"already"`
	Dropped []string  `json:"dropped"`
	Oldest  time.Time `json:"oldest,omitzero"`
	Says    string    `json:"says"`
	Error   string    `json:"error,omitempty"`
}

// bindingsAgain reads module's bindings again from the seat's events the stream holds, keeps each, and writes what it
// found to its own state; an error is said in the pass's errors and in the kept result.
func (p *Part) bindingsAgain(res *Result, module, by string) {
	out := BindingsAgain{V: 1, Module: module, By: by, At: p.now(), Kept: []string{}, Already: []string{}, Dropped: []string{}}
	events, err := p.History.Said()
	if err != nil {
		out.Error = "the shared events stream could not be read: " + err.Error()
	}
	heardCreate := false
	for _, e := range events {
		if out.Oldest.IsZero() || (!e.At.IsZero() && e.At.Before(out.Oldest)) {
			out.Oldest = e.At
		}
		name, ok := brokerEvent(e.Subject, p.Brokers())
		if !ok {
			continue
		}
		var said BrokerEvent
		if json.Unmarshal(e.Data, &said) != nil || said.Module != module {
			continue
		}
		out.Read++
		local := firstOf(said.LocalName, said.Local, said.Binding.Local)
		changed, err := p.HearBroker(e.Subject, e.Data, p.Brokers())
		if err != nil {
			out.Error = fmt.Sprintf("the %s of %s could not be kept: %v", name, local, err)
			break
		}
		switch {
		case name == BrokerDropped && changed:
			out.Dropped = appendOnce(out.Dropped, local)
			out.Kept = slices.DeleteFunc(out.Kept, func(l string) bool { return l == local })
		case name != BrokerDropped:
			heardCreate = true
			out.Dropped = slices.DeleteFunc(out.Dropped, func(l string) bool { return l == local })
			if changed {
				out.Kept = appendOnce(out.Kept, local)
				out.Already = slices.DeleteFunc(out.Already, func(l string) bool { return l == local })
			} else if !slices.Contains(out.Kept, local) {
				out.Already = appendOnce(out.Already, local)
			}
		}
	}
	switch {
	case out.Error != "":
		out.Says = "the re-read of " + module + " stopped: " + out.Error
		res.Errors = append(res.Errors, out.Says)
	case !heardCreate:
		out.Says = fmt.Sprintf("the shared events stream holds no created or recreated of %s (its oldest event: %s): a created older "+
			"than the stream keeps is not there, and its binding comes back only by the seat mesh-broker's list answering each "+
			"entry's binding (a record extending ADR 0595 §7), not by this tool", module, out.Oldest.Format(time.RFC3339))
	default:
		out.Says = fmt.Sprintf("read %d of the seat mesh-broker's events of %s again: kept %v, already kept %v, dropped %v; the pass "+
			"writes them into its memberships", out.Read, module, out.Kept, out.Already, out.Dropped)
	}
	if p.Logf != nil {
		p.Logf("the bindings of %s read again by hand, for %s: %s", module, by, out.Says)
	}
	raw, _ := json.Marshal(out)
	p.changed(res, StateUserList, KeyBindingsAgain, raw)
}

func appendOnce(list []string, s string) []string {
	if slices.Contains(list, s) {
		return list
	}
	return append(list, s)
}

// AskBindingsAgain answers the tool grants_bindings_again: it asks the part's process to read one module's bindings
// again from the seat mesh-broker's events the shared events stream holds, and records the hand act through record (the
// seat mesh-hand-acts) first; a module that is no module's name, or an ask without why, is refused before anything is
// asked. grants_bindings shows what the process found.
func (p *Part) AskBindingsAgain(caller, module, why string, record func(what, why string) error) (any, error) {
	if module == "" || strings.ContainsAny(module, ".*> ") {
		return nil, fmt.Errorf("grants_bindings_again needs one module, by its name: %q is none", module)
	}
	if strings.TrimSpace(why) == "" {
		return nil, errors.New("grants_bindings_again needs why: it is recorded as a hand act")
	}
	if record != nil {
		if err := record(ToolBindingsAgain+" "+module, why); err != nil {
			return nil, fmt.Errorf("the hand act could not be recorded, so nothing was asked: %w", err)
		}
	}
	ask, _ := json.Marshal(Ask{By: caller, At: p.now(), Bindings: module})
	rev, err := p.Store.Put(StateUserList, KeyAsked, ask)
	if err != nil {
		return nil, fmt.Errorf("the ask could not be written: %w", err)
	}
	return map[string]any{"asked": rev, "module": module,
		"says": "the part's process reads the seat mesh-broker's created, recreated and dropped of " + module + " again from the shared " +
			"events stream at this ask, and keeps each as if heard now; grants_bindings shows what it kept, what was already kept, " +
			"and whether the stream still holds its created"}, nil
}

// lastBindingsAgain is the last re-read kept, when it was of module (or any, for an empty module).
func (p *Part) lastBindingsAgain(module string) (*BindingsAgain, error) {
	raw, _, found, err := p.Store.Get(StateUserList, KeyBindingsAgain)
	if err != nil {
		return nil, fmt.Errorf("its own %s %s cannot be read: %w", StateUserList, KeyBindingsAgain, err)
	}
	if !found {
		return nil, nil
	}
	var b BindingsAgain
	if err := json.Unmarshal(raw, &b); err != nil {
		return nil, fmt.Errorf("its own %s %s does not read: %w", StateUserList, KeyBindingsAgain, err)
	}
	if module != "" && b.Module != module {
		return nil, nil
	}
	return &b, nil
}

func sameBinding(a, b membership.BusBinding) bool {
	ra, _ := json.Marshal(a)
	rb, _ := json.Marshal(b)
	return string(ra) == string(rb)
}

// OwnHears are the module streams the part's own membership on node names it hearing (its consumes of another module's
// stream, the bus module's among them), each with its durable consumer: read from its own state memberships, which the
// part writes; none yet is an error, so the hearing is opened at the next change.
func OwnHears(s Store, node string) ([]membership.HeardIssued, error) {
	key := buscompose.MembershipKey(node, Module)
	raw, _, found, err := s.Get(StateMemberships, key)
	if err != nil {
		return nil, fmt.Errorf("its own membership %s cannot be read: %w", key, err)
	}
	if !found {
		return nil, fmt.Errorf("its own membership %s is not composed yet, so the module streams it hears are not known", key)
	}
	bare, err := membership.BareGrantsMembership(raw)
	if err != nil {
		return nil, fmt.Errorf("its own membership %s does not read: %w", key, err)
	}
	var m membership.Membership
	if err := json.Unmarshal(bare, &m); err != nil {
		return nil, fmt.Errorf("its own membership %s does not read: %w", key, err)
	}
	return m.Hears, nil
}

// withBound sets what the bus module said it created on in: the streams of every binding kept (Inputs.Bound) and the
// bindings themselves (Inputs.Bindings), read from the part's own state in one place for the pass and for its tools.
func (p *Part) withBound(in *Inputs) error {
	bound, err := p.boundStreams()
	if err != nil {
		return err
	}
	in.Bound, in.Bindings = bound, map[string][]membership.BusBinding{}
	for module := range bound {
		binds, err := p.bindingsOf(module)
		if err != nil {
			return err
		}
		in.Bindings[module] = binds
	}
	return nil
}

// boundStreams are, by module, the streams on the bus of every binding kept (BoundKey): what the bus module's process
// may still define, an orphan's among them (buscompose.BusStateNames).
func (p *Part) boundStreams() (map[string][]string, error) {
	keys, err := p.Store.Keys(StateMemberships)
	if err != nil {
		return nil, fmt.Errorf("its own %s cannot be listed: %w", StateMemberships, err)
	}
	out := map[string][]string{}
	for _, key := range keys {
		module, ok := strings.CutPrefix(key, BoundPrefix)
		if !ok {
			continue
		}
		b, err := p.bound(module)
		if err != nil {
			return nil, err
		}
		for _, bind := range b.Bindings {
			out[module] = append(out[module], bindingStreams(bind)...)
		}
	}
	return out, nil
}

// bindingStreams are the streams on the bus a binding names: a stream's, a bucket's KV_, an object store's OBJ_, a ledger's
// history, its snapshot stream and its state and index buckets.
func bindingStreams(b membership.BusBinding) []string {
	var out []string
	if b.Stream != "" {
		out = append(out, b.Stream)
	}
	if b.Bucket != "" {
		out = append(out, "KV_"+b.Bucket)
	}
	if b.ObjectStore != "" {
		out = append(out, "OBJ_"+b.ObjectStore)
	}
	if b.History != "" {
		out = append(out, b.History, b.History+"_snapshot")
	}
	if b.State != "" {
		out = append(out, "KV_"+b.State)
	}
	for _, ix := range b.Indexes {
		out = append(out, "KV_"+ix)
	}
	return out
}

func firstOf(values ...string) string {
	for _, v := range values {
		if v != "" {
			return v
		}
	}
	return ""
}