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

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

// 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
	for _, h := range d.Hears {
		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)
}