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

package grants

import (
	"errors"
	"fmt"
	"slices"
	"strings"

	"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
	"git.novox.be/novox/mesh-sdk/go/seatdef"
	"git.novox.be/novox/mesh-sdk/go/views"

	"mesh-grants/internal/buscompose"
)

// **A seat's kept event is heard on its own stream** (novox/hq design 47; buscompose/kept.go): a module that consumes one has
// its durable on the kept stream, from the last message of each subject for an event kept so, and no filter of it on the
// shared events stream.

// SeatDefinitions is what a seat gate may also say: the definitions the seat list holds (seatlist.Reader.Definitions).
type SeatDefinitions interface {
	Definitions() map[string]seatdef.Definition
}

// keptOutOfHearing moves each seat event in watches or in the module's consumes that a kept stream carries out of the shared
// stream's hearing and into hears, one durable on the kept stream each (merged by stream). A watch left with no event is
// dropped. It answers what is left to watch and consume, the hears, and whether any moved.
func keptOutOfHearing(kept []buscompose.KeptEvent, watches []buscompose.Seat, consumes []string, hears []buscompose.Heard) ([]buscompose.Seat, []string, []buscompose.Heard, bool) {
	if len(kept) == 0 {
		return watches, consumes, hears, false
	}
	var left []buscompose.Seat
	var named []string // every consume and every watched event, as `<seat>.<event>`
	named = append(named, consumes...)
	for _, w := range watches {
		var emits []string
		for _, e := range w.Emits {
			if _, ok := buscompose.KeptHeard(kept, w.Name+"."+e); ok {
				named = append(named, w.Name+"."+e)
				continue
			}
			emits = append(emits, e)
		}
		if len(emits) > 0 {
			w.Emits = emits
			left = append(left, w)
		}
	}
	rest, moved := buscompose.SplitKept(kept, consumes)
	_, movedWatched := buscompose.SplitKept(kept, named[len(consumes):])
	moved = append(moved, movedWatched...)
	if len(moved) == 0 {
		return watches, consumes, hears, false
	}
	for _, h := range moved {
		if i := slices.IndexFunc(hears, func(x buscompose.Heard) bool { return x.Stream == h.Stream }); i >= 0 {
			hears[i].Filters = slices.Compact(slices.Sorted(slices.Values(append(slices.Clone(hears[i].Filters), h.Filters...))))
			continue
		}
		hears = append(hears, h)
	}
	return left, rest, hears, true
}

// createKept makes each kept event's stream the bus lacks, from the shared events stream's messages of its subject, and
// says what it made. A stream the bus holds is left as it is; one made at this seat list and gone since is made again only on
// a hand assert (the same rule as a manifest's streams: made once per definition). held is what the bus held at this pass.
func createKept(c Creator, done *creations, kept []buscompose.KeptEvent, held map[string]Limits, force bool) (made, notMade []string) {
	for _, k := range kept {
		if _, ok := held[k.Stream]; ok {
			continue
		}
		cfg := k.StreamConfig()
		digest := fmt.Sprintf("%v", cfg.Sources[0].FilterSubject) + fmt.Sprint(k.Last, k.MaxAgeDays, k.MaxMiB)
		if !force && done.madeAt(k.Stream) == digest {
			notMade = append(notMade, k.Stream+": made at this definition and gone since; grants_assert makes it again")
			continue
		}
		switch err := c.CreateDeclared(cfg); {
		case err == nil:
			done.keep(k.Stream, digest)
			made = append(made, k.Stream)
		case errors.Is(err, ErrHeld):
		default:
			notMade = append(notMade, k.Stream+": "+err.Error()+" (tried again at the next event)")
		}
	}
	return made, notMade
}

// eventSubjectsOf is, by seat, each event's subject pattern (`running.<node>.<module>`) of the events whose subject carries
// parts, by the event's local name (`running.*.*`).
func eventSubjectsOf(defs map[string]seatdef.Definition) map[string]map[string]string {
	out := map[string]map[string]string{}
	for _, d := range defs {
		for _, e := range d.Events {
			if len(e.Parts()) == 0 {
				continue
			}
			if out[d.Seat] == nil {
				out[d.Seat] = map[string]string{}
			}
			out[d.Seat][e.Local()] = e.SubjectPattern()
		}
	}
	return out
}

// definedEventsOf is the events a `*` emitter before them may reach (catalogue.Hearing's defined, design 47 G §386, §474,
// §644): each event, by its local name, of a seat definition of the seat list, and mesh-sdk's three finding events (views.FindingSubjects) and the seat mesh-conditions's findings-asked (its generated
// contract's SubjectFindingsAsked), which are defined with or without the list read.
func definedEventsOf(defs map[string]seatdef.Definition) map[string]bool {
	out := map[string]bool{}
	for _, d := range defs {
		for _, e := range d.Events {
			out[e.Local()] = true
		}
	}
	s := views.FindingSubjects("module", "node")
	for _, subject := range []string{s.Written, s.Cleared, s.Stated} {
		out[strings.TrimSuffix(strings.TrimPrefix(subject, "mesh.stream.module.event."), ".node")] = true
	}
	out[strings.TrimPrefix(meshconditions.SubjectFindingsAsked, "mesh.seat.mesh-conditions.event.")] = true
	return out
}

// withEventSubjects gives each seat a declaration holds or watches the subjects of its events (machineEvent).
func withEventSubjects(subjects map[string]map[string]string, d *buscompose.Declared) {
	if len(subjects) == 0 {
		return
	}
	for _, list := range []*[]buscompose.Seat{&d.Holds, &d.Uses, &d.Watches} {
		for i := range *list {
			(*list)[i].EventSubjects = subjects[(*list)[i].Name]
		}
	}
}

// acceptCallersOf is, by pulled-by-kind seat and then by queue, the seats that may ask the queue: the `seat:<seat>` entries
// of the callers of every accepted kind the definition puts on it (a kind's `stream` is its queue, else its name is). Never a
// name of the grants part's own: it reads the definitions' lists (design 47 G §543, §600, A-163).
func acceptCallersOf(defs map[string]seatdef.Definition) map[string]map[string][]string {
	out := map[string]map[string][]string{}
	for _, d := range defs {
		if !d.PulledByKind {
			continue
		}
		for _, a := range d.Accepts {
			if a.Callers == nil || a.Callers.IsText() {
				continue
			}
			queue := a.Name
			if a.Stream != "" {
				queue = a.Stream
			}
			for _, e := range a.Callers.List {
				if seat, ok := strings.CutPrefix(e, seatdef.CallerSeatPrefix); ok && !slices.Contains(out[d.Seat][queue], seat) {
					if out[d.Seat] == nil {
						out[d.Seat] = map[string][]string{}
					}
					out[d.Seat][queue] = append(out[d.Seat][queue], seat)
				}
			}
		}
	}
	return out
}

// withAcceptCallers gives each pulled-by-kind seat a declaration holds or uses the askers of its queues (Seat.AcceptCallers).
func withAcceptCallers(askers map[string]map[string][]string, d *buscompose.Declared) {
	if len(askers) == 0 {
		return
	}
	for _, list := range []*[]buscompose.Seat{&d.Holds, &d.Uses, &d.Watches} {
		for i := range *list {
			(*list)[i].AcceptCallers = askers[(*list)[i].Name]
		}
	}
}

// withVerbCallers gives each seat a declaration holds, uses or watches the effective callers of its verbs
// (Seat.VerbCallers, design 47 G §610, A-179), as withAcceptCallers gives the askers of its queues.
func withVerbCallers(callers map[string]map[string][]string, d *buscompose.Declared) {
	for _, list := range []*[]buscompose.Seat{&d.Holds, &d.Uses, &d.Watches} {
		for i := range *list {
			(*list)[i].VerbCallers = callers[(*list)[i].Name]
		}
	}
}