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 `.`. 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"}, }