mesh-controller/internal/broker/modstreams.go

package broker

import (
	"fmt"
	"sort"
	"strings"
)

// A module's own stream on the bus (novox/hq ADR 0363).
//
// A module declares a stream as it declares a log: named locally, created by the controller from the
// catalogue on every raise and every send, never removed by the mesh. Its events are published on
// `mesh.stream.<module>.event.<event>`, a subject space the shared EVENTS stream (`mesh.mod.*.event.>`)
// does not take — the bus refuses two streams claiming one subject, so a module's stream cannot share
// EVENTS' — and a consumer hears it through a durable consumer of its own on that stream, named as on
// EVENTS. Pure; the controller's raise asserts what is composed here.

// moduleStreamPrefix starts every subject a module's own stream takes.
const moduleStreamPrefix = "mesh.stream."

// ModuleStreamSubject is where a module's event on its own stream lands: `mesh.stream.<module>.event.<event>`.
// The token before `.event.` is the emitter, as on EVENTS, so a runtime reads the same key from it.
func ModuleStreamSubject(module, event string) string {
	return moduleStreamPrefix + module + ".event." + event
}

// OnAModuleStream says a subject is one a module's own stream takes.
func OnAModuleStream(subject string) bool { return strings.HasPrefix(subject, moduleStreamPrefix) }

// WirePattern is a local event pattern as the bus spells it: `**` is `>`.
func WirePattern(local string) string {
	parts := strings.Split(local, ".")
	for i, p := range parts {
		if p == "**" {
			parts[i] = ">"
		}
	}
	return strings.Join(parts, ".")
}

// A ModuleStream is one module's declared stream as the bus holds it.
type ModuleStream struct {
	Module string
	// Name is its name on the bus, `MAIL`.
	Name string
	// Events are the events it carries, as the bus spells them (WirePattern).
	Events     []string
	MaxAgeDays int
	MaxMiB     int
	// Restricted says a trusted config value of the module grants who hears it (ADR 0363 §4).
	Restricted bool
}

// Subjects are the subjects the stream takes.
func (s ModuleStream) Subjects() []string {
	var out []string
	for _, e := range s.Events {
		out = append(out, ModuleStreamSubject(s.Module, e))
	}
	sort.Strings(out)
	return out
}

// Stream is the stream the controller asserts: kept by limits, its age and size the module's, dropping
// its oldest when full — a consumer that was away catches up within the age, and the module's own record
// answers for anything older.
func (s ModuleStream) Stream() Stream {
	why := fmt.Sprintf("%s's own stream (novox/hq ADR 0363): its events, kept %d days and at most %d MiB so a "+
		"consumer that was away catches up", s.Module, s.MaxAgeDays, s.MaxMiB)
	if s.Restricted {
		why += "; heard only by the consumers its trusted config value grants"
	}
	return Stream{
		Name:      s.Name,
		Subjects:  s.Subjects(),
		Retention: RetentionLimits,
		MaxAge:    s.MaxAgeDays * 24 * 60 * 60,
		MaxBytes:  int64(s.MaxMiB) * 1024 * 1024,
		Why:       why,
		// Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3).
		Restricted: s.Restricted,
	}
}

// ModuleStreams is every declared module stream as the controller asserts it, in name order.
func ModuleStreams(declared []ModuleStream) []Stream {
	sorted := append([]ModuleStream(nil), declared...)
	sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name })
	out := make([]Stream, 0, len(sorted))
	for _, s := range sorted {
		out = append(out, s.Stream())
	}
	return out
}

// Heard is a module's stream another module hears, and the subjects it hears of it: its `consumes`
// within the stream and, for a restricted stream, within what the provider's trusted config value grants it
// (ADR 0363 §5). Composed only with at least one subject.
type Heard struct {
	Stream  string
	Emitter string
	Filters []string
	// Restricted says the stream is restricted by its provider's trusted config value (ADR 0363 §4): then it
	// is heard only by a module on an account of its own (ADR 0376 §1, ComposedHears).
	Restricted bool
}

// ComposedHears is what of a module's hearing the controller composes (novox/hq ADR 0376 §1): every stream it
// hears, but a restricted one only for a module that runs in a runtime of its own on a bus account of its own
// (`runs-as`, ADR 0259). **The tool runner never hears a restricted stream**: a module it carries reaches the bus
// through the tool runner's account, which every agent session on that machine reaches the bus as too, so a
// consumer, its grants or its bind word composed there would hand the stream to any session. A module that
// declares no `runs-as` and keeps its own container credential gets none either: only `runs-as` is an account
// the mesh keeps from agents.
func ComposedHears(d Declared) []Heard {
	if d.RunsAs != "" {
		return d.Hears
	}
	var out []Heard
	for _, h := range d.Hears {
		if !h.Restricted {
			out = append(out, h)
		}
	}
	return out
}

// Withheld is a restricted stream a grant gives a module that the controller composes nothing for, because the
// module has no account of its own (novox/hq ADR 0376 §2): said in `doctor` and in the controller's log, never
// silent.
type Withheld struct {
	Node, Module, Stream, Emitter string
}

// Said is a withheld hearing in the operator's words: the module, the stream and the remedy.
func (w Withheld) Said() string {
	return fmt.Sprintf("%s on %s is granted part of %s's restricted stream %s, and hears none of it: it is carried "+
		"by the machine's tool runner, whose bus account every agent session there shares, so nothing is composed "+
		"for it. Declare runs-as in %s's manifest, so it runs as an account of its own (novox/hq ADR 0376)",
		w.Module, w.Node, w.Emitter, w.Stream, w.Module)
}

// WithheldOf is every granted hearing of a restricted stream composed for nothing, by machine and module.
func WithheldOf(r Records) []Withheld {
	var out []Withheld
	for _, node := range sortedCopy(r.Nodes) {
		for _, d := range r.Assigned[node] {
			if d.RunsAs != "" {
				continue
			}
			for _, h := range d.Hears {
				if h.Restricted && len(h.Filters) > 0 {
					out = append(out, Withheld{Node: node, Module: d.Module, Stream: h.Stream, Emitter: h.Emitter})
				}
			}
		}
	}
	return out
}

// heardConsumer is the durable consumer a module hears a module's stream through: named as its consumer
// on EVENTS, on that stream, with the subjects it was granted as its filters.
func heardConsumer(p Principal, h Heard) Consumer {
	filters := append([]string(nil), h.Filters...)
	sort.Strings(filters)
	return Consumer{
		Name:           consumerDurable(p),
		Stream:         h.Stream,
		Filters:        filters,
		AckWaitSeconds: 30,
		MaxDeliver:     5,
		Why: "what " + p.Module + " consumes of " + h.Emitter + "'s stream " + h.Stream + " and is granted " +
			"(novox/hq ADR 0363); after max-deliver it gives an event up, and the controller keeps it in " +
			"DEAD_LETTERS until a person delivers it again or drops it",
	}
}

// heardGrants are what a principal publishes to read and acknowledge the consumers it hears module streams
// through: asking about each, pulling from it, acknowledging it — its own name on each stream, no other's.
func heardGrants(p Principal, hears []Heard) []string {
	var out []string
	durable := consumerDurable(p)
	for _, h := range hears {
		if !safeSubject.MatchString(h.Stream) || len(h.Filters) == 0 {
			continue
		}
		out = append(out,
			"$JS.API.CONSUMER.INFO."+h.Stream+"."+durable,
			"$JS.API.CONSUMER.MSG.NEXT."+h.Stream+"."+durable,
			"$JS.ACK."+h.Stream+"."+durable+".>")
	}
	return out
}

// ownStreamGrants are the subjects a module publishes its own streams' events on, and no other's.
func ownStreamGrants(module string, streams []ModuleStream) []string {
	var out []string
	for _, s := range streams {
		if s.Module != module {
			continue
		}
		out = append(out, s.Subjects()...)
	}
	return out
}

// StreamIssued is one of a module's own streams as its membership names it (ADR 0363 §2): the events it
// carries, as the bus spells them, and the subject an event of them is published on, `{event}` the event.
type StreamIssued struct {
	Stream  string   `json:"stream"`
	Events  []string `json:"events"`
	Subject string   `json:"subject"`
}

// HeardIssued is one module stream an assignment hears, as its membership names it (ADR 0363 §8): the
// stream and the durable consumer on it the runtime binds.
type HeardIssued struct {
	Stream   string `json:"stream"`
	Consumer string `json:"consumer"`
}

// streamsIssuedFor is every stream of its own a module's membership names.
func streamsIssuedFor(d Declared) []StreamIssued {
	var out []StreamIssued
	for _, s := range d.Streams {
		if s.Module != d.Module || !safeSubject.MatchString(d.Module) {
			continue
		}
		out = append(out, StreamIssued{Stream: s.Name, Events: append([]string(nil), s.Events...),
			Subject: ModuleStreamSubject(d.Module, "{event}")})
	}
	return out
}

// HeardIssuedFor is every module stream an assignment hears, as its membership names it.
func HeardIssuedFor(node string, d Declared) []HeardIssued {
	var out []HeardIssued
	// A restricted stream's bind word only for a module of its own account (novox/hq ADR 0376 §1).
	for _, h := range ComposedHears(d) {
		if len(h.Filters) == 0 {
			continue
		}
		out = append(out, HeardIssued{Stream: h.Stream, Consumer: node + "_" + d.Module})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Stream < out[j].Stream })
	return out
}

// Pruner is the part of a JetStream connection that removes a module's consumer on a module's stream it
// may no longer hear (ADR 0363 §6). Optional: what only records what it is asked to assert has none.
type Pruner interface {
	ConsumerNames(stream string) ([]string, error)
	RemoveConsumer(stream, name string) error
}

// PruneHeard removes, from each module stream, every durable consumer the composition no longer asks for,
// and answers what it removed as `<stream>/<consumer>`. A consumer that loses its grant loses what it reads
// through, whatever its runtime still holds; a stream that does not exist yet has nothing to remove.
func PruneHeard(p Pruner, streams []ModuleStream, wanted []Consumer) ([]string, error) {
	keep := map[string]bool{}
	for _, c := range wanted {
		keep[c.Stream+"/"+c.Name] = true
	}
	var removed []string
	for _, s := range streams {
		names, err := p.ConsumerNames(s.Name)
		if err != nil {
			return removed, fmt.Errorf("listing the consumers on %s: %w", s.Name, err)
		}
		for _, n := range names {
			if keep[s.Name+"/"+n] {
				continue
			}
			if err := p.RemoveConsumer(s.Name, n); err != nil {
				return removed, fmt.Errorf("removing %s's consumer on %s, which is no longer granted: %w", n, s.Name, err)
			}
			removed = append(removed, s.Name+"/"+n)
		}
	}
	return removed, nil
}

// EnsureHeard asserts a consumer on a module's stream (ADR 0363 §5, §6). **On a restricted stream, a consumer
// the bus will not bring to match is removed**: a pull consumer delivers to whoever pulls it, so one left with
// its wider filters after a narrowing was refused would keep handing over what the grant no longer names,
// whatever the user list now says. The next assertion makes it again with the granted filters. The refusal is
// returned either way, saying what was done, so the send says it and raises it.
func EnsureHeard(r Raiser, c Consumer, restricted bool) error {
	err := r.EnsureConsumer(c)
	if err == nil || !restricted {
		return err
	}
	p, prunes := r.(Pruner)
	if !prunes {
		return fmt.Errorf("%w; it could not be removed here, and may still deliver more than it is granted", err)
	}
	if rerr := p.RemoveConsumer(c.Stream, c.Name); rerr != nil {
		return fmt.Errorf("%w; REMOVING IT FAILED TOO, so it may still deliver more than it is granted: %v", err, rerr)
	}
	return fmt.Errorf("%w; REMOVED, so it delivers nothing it is no longer granted: the next assertion makes it "+
		"again with its granted filters (novox/hq ADR 0363 §6)", err)
}