384 lines
13 KiB
Go
384 lines
13 KiB
Go
package catalogue
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"slices"
|
|
"strings"
|
|
)
|
|
|
|
// A module's own stream (novox/hq ADR 0363).
|
|
//
|
|
// A module declares a stream as it declares a log (ADR 0297) or a bucket (ADR 0201): named locally, the
|
|
// events it carries named locally as `emits` names them, its age and its size within the mesh's bounds.
|
|
// The controller creates it and stays the only writer of stream definitions. A stream may be restricted
|
|
// by a trusted config value of its own module: then a consumer hears a part of it only where its own
|
|
// `consumes` asks and that value names it.
|
|
|
|
// The mesh's bounds on a module's stream: its age in days and its size in MiB, and what it gets when it
|
|
// says nothing.
|
|
const (
|
|
StreamLeastDays = 1
|
|
StreamMostDays = 30
|
|
StreamDefaultDays = 7
|
|
StreamLeastMiB = 1
|
|
StreamMostMiB = 8192
|
|
StreamDefaultMiB = 1024
|
|
)
|
|
|
|
// GrantKey is the name in a grant's pattern that each key of the config value fills.
|
|
const GrantKey = "{key}"
|
|
|
|
// StreamRetentionWorkQueue is a stream's declared retention (novox/hq ADR 0436 §6): a message is removed
|
|
// once a consumer acknowledges it, and the bus refuses a new one rather than dropping what it holds. Empty
|
|
// is the mesh's own limits retention — kept until it ages or the stream fills, dropping its oldest.
|
|
const StreamRetentionWorkQueue = "work-queue"
|
|
|
|
// StreamLeastDeliveries and StreamMostDeliveries bound a work-queue stream's max-deliveries (novox/hq ADR
|
|
// 0436 §6): how many times a message is handed over before it is given up as a dead letter.
|
|
const (
|
|
StreamLeastDeliveries = 1
|
|
StreamMostDeliveries = 10
|
|
)
|
|
|
|
|
|
// MeshStreamNames are the mesh's own streams by name. Written here as well as in the broker, which this
|
|
// package cannot import; a test keeps them agreeing.
|
|
var MeshStreamNames = []string{"CONTROL", "NODES", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS", "PROV_APPLIED", "CHECKS_READY", "CHECKS_OTHER"}
|
|
|
|
// StreamDeclaration is one stream a module owns.
|
|
type StreamDeclaration struct {
|
|
Name string `json:"name"`
|
|
// Events are the events it carries, named locally; `*` is one name and `**`, last only, the rest.
|
|
Events []string `json:"events"`
|
|
// MaxAgeDays and MaxMiB bound it; zero is the default.
|
|
MaxAgeDays int `json:"max-age-days,omitempty"`
|
|
MaxMiB int `json:"max-mib,omitempty"`
|
|
// Grant restricts it, when said.
|
|
Grant *StreamGrant `json:"grant,omitempty"`
|
|
// Retention is how the stream decides what to keep when it fills (novox/hq ADR 0436 §6): empty for the
|
|
// mesh's own limits retention (dropping its oldest), or StreamRetentionWorkQueue.
|
|
Retention string `json:"retention,omitempty"`
|
|
// MaxDeliveries bounds how many times a work-queue message is handed over, 1 to 10; refused without
|
|
// retention work-queue (novox/hq ADR 0436 §6).
|
|
MaxDeliveries int `json:"max-deliveries,omitempty"`
|
|
// Duplicates is a work-queue stream's duplicate window in hours, at most its maximum age; refused
|
|
// without retention work-queue (novox/hq ADR 0436 §6).
|
|
Duplicates int `json:"duplicates,omitempty"`
|
|
}
|
|
|
|
// StreamGrant says which trusted config value of the module grants a consumer a part of the stream, and
|
|
// which part each of its keys names.
|
|
type StreamGrant struct {
|
|
// Config is the trusted config value of the module that grants it (ADR 0348 §8): set only at the
|
|
// controller's terminal or on the operator's Approve.
|
|
Config string `json:"config"`
|
|
// Events is a pattern holding GrantKey exactly once, which each key of the config value fills.
|
|
Events string `json:"events"`
|
|
}
|
|
|
|
// UnmarshalJSON reads a stream strictly: an unknown key is refused here, not passed over.
|
|
func (s *StreamDeclaration) UnmarshalJSON(raw []byte) error {
|
|
type plain StreamDeclaration
|
|
var p plain
|
|
dec := json.NewDecoder(bytes.NewReader(raw))
|
|
dec.DisallowUnknownFields()
|
|
if err := dec.Decode(&p); err != nil {
|
|
return fmt.Errorf("a stream is {name, events, max-age-days, max-mib, grant, retention, max-deliveries, "+
|
|
"duplicates}: %w", typedUnknown(err))
|
|
}
|
|
*s = StreamDeclaration(p)
|
|
return nil
|
|
}
|
|
|
|
// UnmarshalJSON reads a grant strictly.
|
|
func (g *StreamGrant) UnmarshalJSON(raw []byte) error {
|
|
type plain StreamGrant
|
|
var p plain
|
|
dec := json.NewDecoder(bytes.NewReader(raw))
|
|
dec.DisallowUnknownFields()
|
|
if err := dec.Decode(&p); err != nil {
|
|
return fmt.Errorf("a stream's grant is {config, events}: %w", typedUnknown(err))
|
|
}
|
|
*g = StreamGrant(p)
|
|
return nil
|
|
}
|
|
|
|
// Days is the stream's age in days, the default where none is said.
|
|
func (s StreamDeclaration) Days() int {
|
|
if s.MaxAgeDays == 0 {
|
|
return StreamDefaultDays
|
|
}
|
|
return s.MaxAgeDays
|
|
}
|
|
|
|
// Cap is the stream's size in MiB, the default where none is said.
|
|
func (s StreamDeclaration) Cap() int {
|
|
if s.MaxMiB == 0 {
|
|
return StreamDefaultMiB
|
|
}
|
|
return s.MaxMiB
|
|
}
|
|
|
|
// BusName is the stream's name on the bus: its local name in capitals, a hyphen an underscore.
|
|
func (s StreamDeclaration) BusName() string { return StreamBusName(s.Name) }
|
|
|
|
// StreamBusName is a local stream name as the bus holds it.
|
|
func StreamBusName(local string) string {
|
|
return strings.ToUpper(strings.ReplaceAll(local, "-", "_"))
|
|
}
|
|
|
|
// Carries says whether an event, or every event a pattern names, is one this stream carries: within one
|
|
// of its patterns.
|
|
func (s StreamDeclaration) Carries(pattern string) bool {
|
|
for _, p := range s.Events {
|
|
if within(pattern, p) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// StreamsHearing is every stream of a module that carries some of a consumed pattern's events, and
|
|
// whether one of them carries all of them.
|
|
func (m Manifest) StreamsHearing(pattern string) (hearing []StreamDeclaration, whole bool) {
|
|
for _, s := range m.Streams {
|
|
for _, p := range s.Events {
|
|
if patternsOverlap(pattern, p) {
|
|
hearing = append(hearing, s)
|
|
break
|
|
}
|
|
}
|
|
if s.Carries(pattern) {
|
|
whole = true
|
|
}
|
|
}
|
|
return hearing, whole
|
|
}
|
|
|
|
// within says every event pattern a names is one b names too.
|
|
func within(a, b string) bool {
|
|
x, y := strings.Split(a, "."), strings.Split(b, ".")
|
|
for i := 0; ; i++ {
|
|
switch {
|
|
case i == len(y):
|
|
return i == len(x)
|
|
case y[i] == TheRest:
|
|
return i < len(x)
|
|
case i == len(x):
|
|
return false
|
|
case x[i] == TheRest:
|
|
return false
|
|
case y[i] == OneName:
|
|
continue
|
|
case x[i] == OneName:
|
|
return false
|
|
case x[i] != y[i]:
|
|
return false
|
|
}
|
|
}
|
|
}
|
|
|
|
// patternsOverlap says some event both patterns name.
|
|
func patternsOverlap(a, b string) bool {
|
|
x, y := strings.Split(a, "."), strings.Split(b, ".")
|
|
for i := 0; ; i++ {
|
|
switch {
|
|
case i == len(x) && i == len(y):
|
|
return true
|
|
case i == len(x) || i == len(y):
|
|
return false
|
|
case x[i] == TheRest || y[i] == TheRest:
|
|
return true
|
|
case x[i] == OneName || y[i] == OneName || x[i] == y[i]:
|
|
continue
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
}
|
|
|
|
// GrantedKeys reads a restricting config value (novox/hq ADR 0363 §4): each key whose list names the
|
|
// module wherever it runs, or the module on this machine as `<machine>.<module>`. A value of another shape
|
|
// grants nothing; a key that is not one usable name grants nothing.
|
|
func GrantedKeys(value any, module, machine string) []string {
|
|
if text, isText := value.(string); isText {
|
|
// A layer may hold the object as its JSON text.
|
|
var parsed any
|
|
if json.Unmarshal([]byte(text), &parsed) != nil {
|
|
return nil
|
|
}
|
|
value = parsed
|
|
}
|
|
entries, ok := value.(map[string]any)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
var out []string
|
|
for key, listed := range entries {
|
|
if !eventName.MatchString(key) {
|
|
continue
|
|
}
|
|
names, ok := listed.([]any)
|
|
if !ok {
|
|
continue
|
|
}
|
|
for _, n := range names {
|
|
s, _ := n.(string)
|
|
if s == module || (machine != "" && s == machine+"."+module) {
|
|
out = append(out, key)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
slices.Sort(out)
|
|
return out
|
|
}
|
|
|
|
// Granted is a grant's pattern with its key filled.
|
|
func (g StreamGrant) Granted(key string) string { return strings.Replace(g.Events, GrantKey, key, 1) }
|
|
|
|
// narrow is the events both patterns name, as one pattern; false when they name none in common.
|
|
func narrow(a, b string) (string, bool) {
|
|
x, y := strings.Split(a, "."), strings.Split(b, ".")
|
|
var out []string
|
|
for i := 0; ; i++ {
|
|
switch {
|
|
case i == len(x) && i == len(y):
|
|
return strings.Join(out, "."), true
|
|
case i == len(x) || i == len(y):
|
|
return "", false
|
|
case x[i] == TheRest:
|
|
return strings.Join(append(out, y[i:]...), "."), true
|
|
case y[i] == TheRest:
|
|
return strings.Join(append(out, x[i:]...), "."), true
|
|
case x[i] == OneName:
|
|
out = append(out, y[i])
|
|
case y[i] == OneName || x[i] == y[i]:
|
|
out = append(out, x[i])
|
|
default:
|
|
return "", false
|
|
}
|
|
}
|
|
}
|
|
|
|
// HeardPart is what a consumer hears of one module's stream: the patterns, local, it is granted.
|
|
type HeardPart struct {
|
|
Emitter string
|
|
Stream StreamDeclaration
|
|
Patterns []string
|
|
}
|
|
|
|
// Hearing is what a consumer's `consumes` hears of the module streams the shelf declares, on one machine
|
|
// (novox/hq ADR 0363 §5): for each stream, its `consumes` narrowed to the stream's events and, for a
|
|
// restricted stream, to the parts the provider's trusted config value grants it. values answers a restricting config
|
|
// value on each machine running the provider. Answered too: whether any `consumes` names a module owning a
|
|
// stream, granted or not, and the `consumes` left for EVENTS — those naming something the provider emits
|
|
// there.
|
|
//
|
|
// Only an entry naming its emitter literally reaches a module's stream: a wildcard emitter, `**` among them,
|
|
// is EVENTS' alone, so nothing that hears every event hears a restricted stream by saying so.
|
|
func Hearing(consumer Manifest, shelf map[string]Manifest, machine string,
|
|
values func(provider, setting string) []any) (parts []HeardPart, touches bool, onEvents []string) {
|
|
byStream := map[string]*HeardPart{}
|
|
var order []string
|
|
for _, c := range consumer.Consumes {
|
|
emitter, rest, named := strings.Cut(c, ".")
|
|
provider, known := shelf[emitter]
|
|
if !named || emitter == OneName || emitter == TheRest || !known || len(provider.Streams) == 0 {
|
|
onEvents = append(onEvents, c)
|
|
continue
|
|
}
|
|
hearing, whole := provider.StreamsHearing(rest)
|
|
if len(hearing) > 0 {
|
|
touches = true
|
|
}
|
|
// Kept for EVENTS while the provider says something there it names: a stream carrying it whole, or
|
|
// a provider that says nothing of it on EVENTS, leaves it nothing to hear there.
|
|
if !whole && slices.ContainsFunc(provider.EmitsAll(), func(e string) bool { return patternsOverlap(rest, e) }) {
|
|
onEvents = append(onEvents, c)
|
|
}
|
|
for _, s := range hearing {
|
|
var granted []string
|
|
if s.Grant != nil {
|
|
for _, v := range values(emitter, s.Grant.Config) {
|
|
for _, key := range GrantedKeys(v, consumer.Module, machine) {
|
|
granted = append(granted, s.Grant.Granted(key))
|
|
}
|
|
}
|
|
}
|
|
var got []string
|
|
for _, p := range s.Events {
|
|
x, ok := narrow(rest, p)
|
|
if !ok {
|
|
continue
|
|
}
|
|
if s.Grant == nil {
|
|
got = append(got, x)
|
|
continue
|
|
}
|
|
for _, g := range granted {
|
|
if y, ok := narrow(x, g); ok {
|
|
got = append(got, y)
|
|
}
|
|
}
|
|
}
|
|
if len(got) == 0 {
|
|
continue
|
|
}
|
|
id := emitter + "/" + s.BusName()
|
|
if byStream[id] == nil {
|
|
byStream[id] = &HeardPart{Emitter: emitter, Stream: s}
|
|
order = append(order, id)
|
|
}
|
|
byStream[id].Patterns = append(byStream[id].Patterns, got...)
|
|
}
|
|
}
|
|
for _, id := range order {
|
|
h := byStream[id]
|
|
h.Patterns = uncovered(h.Patterns)
|
|
parts = append(parts, *h)
|
|
}
|
|
return parts, touches, onEvents
|
|
}
|
|
|
|
// uncovered is a list of patterns without those another names whole, and without repeats: the bus refuses
|
|
// a consumer whose filters overlap.
|
|
func uncovered(patterns []string) []string {
|
|
var out []string
|
|
for i, p := range patterns {
|
|
covered := false
|
|
for j, q := range patterns {
|
|
if i == j {
|
|
continue
|
|
}
|
|
if within(p, q) && (!within(q, p) || j < i) {
|
|
covered = true
|
|
break
|
|
}
|
|
}
|
|
if !covered {
|
|
out = append(out, p)
|
|
}
|
|
}
|
|
slices.Sort(out)
|
|
return out
|
|
}
|
|
|
|
// **A module keeps its data in the bus's predefined forms** (the lead and the operator, 2026-10-29; the operator's rule of
|
|
// 2026-10-22, "data storage is done using our predefined nats bus options"): a state, a log or an object store, which the
|
|
// runtime reads and writes for its tools and other modules read by a declared `reads`. A stream of its own is outside
|
|
// them and refused, naming it. StreamsKeptUntilMoved are the streams declared before this check, each allowed by name
|
|
// until its module moves it; the list only shrinks.
|
|
var StreamsKeptUntilMoved = map[string][]string{
|
|
"brother": {"device-asks"},
|
|
"mesh-build-go": {"toolchains-go"},
|
|
"mesh-build-node": {"toolchains-node"},
|
|
"mesh-mail": {"mail", "mail-sends"},
|
|
"only-office": {"convert-asks"},
|
|
"mesh-delivery-assigner": {"placement"},
|
|
"mesh-delivery-checker": {"checks"},
|
|
"mesh-delivery-forge-bridge": {"forge"},
|
|
}
|