mesh-controller/internal/catalogue/streams.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 setting's value fills.
const GrantKey = "{key}"

// ReservedStreamPrefixes are the starts of the stream names the mesh derives, which no module's stream
// may take: a seat's work queue, a log, a bucket, an object store, and the dead letters.
var ReservedStreamPrefixes = []string{"SEAT_", "LOG_", "KV_", "OBJ_", "DEAD_"}

// 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", "ASSIGNMENTS", "EVENTS", "DEAD_LETTER_NOTICES", "DEAD_LETTERS"}

// 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"`
}

// StreamGrant says which trusted setting 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 setting's 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}: %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
}

// StreamProblems is what is wrong with a manifest's streams, judged on the one manifest.
func StreamProblems(m Manifest) []string {
	if len(m.Streams) == 0 {
		return nil
	}
	var problems []string
	if !stateName.MatchString(m.Module) {
		problems = append(problems, fmt.Sprintf(
			"%s declares a stream, and a module's name is part of its stream's subjects, which take one plain "+
				"name — no dot (novox/hq ADR 0363)", m.Module))
	}
	seen := map[string]bool{}
	terminal := TerminalKeys(m)
	for i, s := range m.Streams {
		switch {
		case !stateName.MatchString(s.Name):
			problems = append(problems, fmt.Sprintf(
				"%s declares stream %q: a stream is named locally — lower-case letters, digits and hyphens "+
					"(novox/hq ADR 0363)", m.Module, s.Name))
		case seen[s.BusName()]:
			problems = append(problems, fmt.Sprintf("%s declares stream %q twice", m.Module, s.Name))
		case slices.Contains(MeshStreamNames, s.BusName()):
			problems = append(problems, fmt.Sprintf(
				"%s declares stream %q, which is the mesh's own stream %s (novox/hq ADR 0363)", m.Module, s.Name, s.BusName()))
		default:
			for _, prefix := range ReservedStreamPrefixes {
				if strings.HasPrefix(s.BusName(), prefix) {
					problems = append(problems, fmt.Sprintf(
						"%s declares stream %q, whose name on the bus starts with %s, which the mesh derives names "+
							"with (novox/hq ADR 0363)", m.Module, s.Name, prefix))
				}
			}
		}
		seen[s.BusName()] = true
		if s.MaxAgeDays != 0 && (s.MaxAgeDays < StreamLeastDays || s.MaxAgeDays > StreamMostDays) {
			problems = append(problems, fmt.Sprintf("%s keeps stream %q %d days; a stream keeps between %d and %d "+
				"(novox/hq ADR 0363)", m.Module, s.Name, s.MaxAgeDays, StreamLeastDays, StreamMostDays))
		}
		if s.MaxMiB != 0 && (s.MaxMiB < StreamLeastMiB || s.MaxMiB > StreamMostMiB) {
			problems = append(problems, fmt.Sprintf("%s caps stream %q at %d MiB; a stream holds between %d and %d "+
				"(novox/hq ADR 0363)", m.Module, s.Name, s.MaxMiB, StreamLeastMiB, StreamMostMiB))
		}
		if len(s.Events) == 0 {
			problems = append(problems, fmt.Sprintf("%s declares stream %q with no events", m.Module, s.Name))
		}
		for _, e := range s.Events {
			if err := eventPattern(e, false); err != nil {
				problems = append(problems, fmt.Sprintf("%s's stream %q carries %q: %v", m.Module, s.Name, e, err))
			}
		}
		// **An event goes to one place** (ADR 0363 §2): an emitted event a stream carries, or two streams
		// carrying one event, would leave the module's runtime to choose.
		for _, e := range m.EmitsAll() {
			if s.Carries(e) {
				problems = append(problems, fmt.Sprintf(
					"%s emits %q, which its stream %q carries: an event goes to EVENTS or to the module's stream, "+
						"so name it in one place (novox/hq ADR 0363)", m.Module, e, s.Name))
			}
		}
		for _, other := range m.Streams[i+1:] {
			for _, a := range s.Events {
				for _, b := range other.Events {
					if patternsOverlap(a, b) {
						problems = append(problems, fmt.Sprintf(
							"%s's streams %q and %q both carry %q and %q: an event goes to one stream (novox/hq "+
								"ADR 0363)", m.Module, s.Name, other.Name, a, b))
					}
				}
			}
		}
		if g := s.Grant; g != nil {
			if !slices.Contains(terminal, g.Config) {
				problems = append(problems, fmt.Sprintf(
					"%s's stream %q is granted by %q, which is not a trusted config value of the module, set only at "+
						"the controller's terminal or on the operator's Approve: a grant of what a module hears is the "+
						"operator's (novox/hq ADR 0348 §8, ADR 0363)", m.Module, s.Name, g.Config))
			}
			if strings.Count(g.Events, GrantKey) != 1 {
				problems = append(problems, fmt.Sprintf(
					"%s's stream %q grants %q, which must hold %s exactly once, as a name of its own (novox/hq ADR 0363)",
					m.Module, s.Name, g.Events, GrantKey))
			} else if err := eventPattern(g.Events, true); err != nil {
				problems = append(problems, fmt.Sprintf("%s's stream %q grants %q: %v", m.Module, s.Name, g.Events, err))
			}
		}
	}
	return problems
}

// eventPattern checks one local event pattern: names, `*` for one, `**` last for the rest; with key,
// GrantKey is a name too.
func eventPattern(pattern string, key bool) error {
	if pattern == "" {
		return fmt.Errorf("a pattern needs a name")
	}
	parts := strings.Split(pattern, ".")
	for i, part := range parts {
		switch {
		case part == TheRest:
			if i != len(parts)-1 {
				return fmt.Errorf("%q stands for the rest of a name, so nothing may follow it", TheRest)
			}
		case part == OneName:
		case key && part == GrantKey:
		case !eventName.MatchString(part):
			return fmt.Errorf("%q is not a usable name: lower-case letters, digits and dashes", part)
		}
	}
	return nil
}

// 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
		}
	}
}

// StreamNameProblems are the rules about stream names no single manifest can be judged against: two
// modules declaring one name.
func StreamNameProblems(shelf Shelf) []string {
	var problems []string
	declaredBy := map[string]string{}
	for _, module := range shelfOrder(shelf) {
		for _, s := range shelf[module].Streams {
			name := s.BusName()
			if first, taken := declaredBy[name]; taken && first != module {
				problems = append(problems, fmt.Sprintf(
					"%s declares the stream %s, which %s already declares; a stream has one owner (novox/hq ADR 0363)",
					module, name, first))
				continue
			}
			declaredBy[name] = module
		}
	}
	return problems
}

// 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 setting grants it. values answers a restricting setting's
// 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
}