mesh-controller/internal/broker/modstreams_test.go

package broker

import (
	"errors"
	"reflect"
	"slices"
	"strings"
	"testing"
)

// mailStream is mesh-mail's own stream as the bus holds it (novox/hq ADR 0363).
var mailStream = ModuleStream{Module: "mesh-mail", Name: "MAIL",
	Events:     []string{"mail.*.message.added", "mail.*.message.changed", "mail.*.message.removed"},
	MaxAgeDays: 7, MaxMiB: 1024, Restricted: true}

// mailHeard is what a reader granted the account `mesh` hears of it.
var mailHeard = Heard{Stream: "MAIL", Emitter: "mesh-mail", Filters: []string{
	"mesh.stream.mesh-mail.event.mail.mesh.message.added", "mesh.stream.mesh-mail.event.mail.mesh.message.changed"}}

// A module's stream is its own: subjects EVENTS does not claim, its age and size, kept by limits (ADR 0363 §1, §2).
func TestAModuleStreamTakesSubjectsEventsDoesNot(t *testing.T) {
	s := mailStream.Stream()
	if s.Name != "MAIL" || s.Retention != RetentionLimits || s.MaxAge != 7*24*60*60 || s.MaxBytes != 1024<<20 || s.DiscardNew {
		t.Fatalf("composed as %+v", s)
	}
	want := []string{"mesh.stream.mesh-mail.event.mail.*.message.added", "mesh.stream.mesh-mail.event.mail.*.message.changed",
		"mesh.stream.mesh-mail.event.mail.*.message.removed"}
	if !reflect.DeepEqual(s.Subjects, want) {
		t.Fatalf("subjects %v", s.Subjects)
	}
	for _, own := range MeshStreams() {
		for _, a := range own.Subjects {
			for _, b := range s.Subjects {
				if SubjectsOverlap(a, b) {
					t.Fatalf("%s claims %s, which overlaps %s of %s: the bus would refuse the stream", s.Name, b, a, own.Name)
				}
			}
		}
	}
	if WirePattern("mail.**") != "mail.>" {
		t.Fatal("`**` is spelled `>` on the wire")
	}
}

// The provider publishes its stream's events on their subjects and nowhere on EVENTS for them; a consumer
// granted a part hears it through its own consumer on the stream, and one not granted has nothing there
// (ADR 0363 §2, §5).
func TestTheProviderPublishesItsStreamAndOnlyAGrantedConsumerHearsIt(t *testing.T) {
	provider, err := PermissionsFor(Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail",
		Streams: []ModuleStream{mailStream}})
	if err != nil {
		t.Fatal(err)
	}
	if !MayPublish(provider, "mesh.stream.mesh-mail.event.mail.work.message.added") {
		t.Fatal("the provider may not publish its own stream's event")
	}
	if MayPublish(provider, "mesh.mod.mesh-mail.event.mail.work.message.added") {
		t.Fatal("the provider may publish its stream's event on EVENTS")
	}
	if MayPublish(provider, "mesh.stream.other.event.mail.work.message.added") {
		t.Fatal("the provider may publish on another module's stream")
	}

	reader := Principal{Kind: KindModule, Node: "ace", Module: "reader", Hears: []Heard{mailHeard}}
	perms, err := PermissionsFor(reader)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range []string{"$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader", "$JS.API.CONSUMER.INFO.MAIL.ace_reader",
		"$JS.ACK.MAIL.ace_reader.1"} {
		if !MayPublish(perms, s) {
			t.Errorf("the granted reader may not publish %s", s)
		}
	}
	if !MaySubscribe(perms, "mesh.stream.mesh-mail.event.mail.mesh.message.added") ||
		MaySubscribe(perms, "mesh.stream.mesh-mail.event.mail.work.message.added") {
		t.Fatal("the reader's subscription is not what it was granted")
	}
	if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_other") {
		t.Fatal("the reader may pull another's consumer")
	}

	stranger, err := PermissionsFor(Principal{Kind: KindModule, Node: "ace", Module: "stranger",
		Consumes: []string{"billing.paid"}})
	if err != nil {
		t.Fatal(err)
	}
	if MayPublish(stranger, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") ||
		MaySubscribe(stranger, "mesh.stream.mesh-mail.event.mail.mesh.message.added") {
		t.Fatal("a module not granted reaches the stream")
	}
}

// A granted consumer has a durable consumer on the stream, with its granted subjects as filters, and an EVENTS
// consumer that hears what it gives up on delivered again (ADR 0363 §5, §7); the runtime carrying it may read both.
func TestAGrantedConsumerHasAConsumerOnTheStream(t *testing.T) {
	records := Records{Nodes: []string{"ace", "novox"}, Assigned: map[string][]Declared{
		"novox": {{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}},
		"ace":   {{Module: RuntimeModule}, {Module: "reader", Hears: []Heard{mailHeard}, HearsAStream: true}, {Module: "stranger", HearsAStream: true}},
	}}
	users, err := Users(records)
	if err != nil {
		t.Fatal(err)
	}
	var onMail, onEvents []Consumer
	for _, c := range ConsumersOf(users) {
		switch c.Consumer.Stream {
		case "MAIL":
			onMail = append(onMail, c.Consumer)
		case EventsStream:
			onEvents = append(onEvents, c.Consumer)
		}
	}
	if len(onMail) != 1 || onMail[0].Name != "ace_reader" || !slices.Equal(onMail[0].Filters, mailHeard.Filters) {
		t.Fatalf("on MAIL: %+v", onMail)
	}
	if len(onEvents) != 1 || onEvents[0].Name != "ace_reader" || !slices.Equal(onEvents[0].Filters, []string{AgainFilter("ace_reader")}) {
		t.Fatalf("on EVENTS: %+v", onEvents)
	}
	for _, u := range users {
		if u.Kind != KindNodeTools || u.Node != "ace" {
			continue
		}
		perms, err := PermissionsFor(u)
		if err != nil {
			t.Fatal(err)
		}
		if !MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_reader") || !MayPublish(perms, "$JS.ACK.MAIL.ace_reader.1") {
			t.Fatal("the runtime carrying the reader may not read its consumer on MAIL")
		}
		if MayPublish(perms, "$JS.API.CONSUMER.MSG.NEXT.MAIL.ace_stranger") {
			t.Fatal("the runtime may read a consumer no module it carries is granted")
		}
	}
}

// The membership says where a module's stream events go and which consumers on module streams to bind
// (ADR 0363 §2, §8).
func TestTheMembershipNamesTheStreamAndWhatIsHeard(t *testing.T) {
	where := Placements{Nodes: map[string][]string{}}
	provider := MembershipFor("novox", Declared{Module: "mesh-mail", Streams: []ModuleStream{mailStream}}, where)
	if len(provider.Streams) != 1 || provider.Streams[0].Stream != "MAIL" ||
		provider.Streams[0].Subject != "mesh.stream.mesh-mail.event.{event}" ||
		!slices.Equal(provider.Streams[0].Events, mailStream.Events) {
		t.Fatalf("the provider's membership: %+v", provider.Streams)
	}
	reader := MembershipFor("ace", Declared{Module: "reader", Hears: []Heard{mailHeard}}, where)
	if !reflect.DeepEqual(reader.Hears, []HeardIssued{{Stream: "MAIL", Consumer: "ace_reader"}}) {
		t.Fatalf("the reader's membership: %+v", reader.Hears)
	}
	if none := MembershipFor("ace", Declared{Module: "stranger", HearsAStream: true}, where); none.Hears != nil {
		t.Fatalf("a module not granted is told to bind %+v", none.Hears)
	}
}

type fakePruner struct {
	on      map[string][]string
	removed []string
}

func (f *fakePruner) ConsumerNames(stream string) ([]string, error) { return f.on[stream], nil }
func (f *fakePruner) RemoveConsumer(stream, name string) error {
	f.removed = append(f.removed, stream+"/"+name)
	return nil
}

// A consumer that lost its grant has its consumer on the stream removed by the next assertion (ADR 0363 §6).
func TestAConsumerThatLostItsGrantIsRemoved(t *testing.T) {
	f := &fakePruner{on: map[string][]string{"MAIL": {"ace_reader", "ace_former"}}}
	wanted := []Consumer{{Name: "ace_reader", Stream: "MAIL"}, {Name: "ace_former", Stream: EventsStream}}
	removed, err := PruneHeard(f, []ModuleStream{mailStream}, wanted)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Equal(removed, []string{"MAIL/ace_former"}) || !slices.Equal(f.removed, removed) {
		t.Fatalf("removed %v", removed)
	}
}

type refusingRaiser struct {
	fakePruner
	refuse bool
}

func (r *refusingRaiser) EnsureStream(Stream) error { return nil }
func (r *refusingRaiser) EnsureConsumer(Consumer) error {
	if r.refuse {
		return errors.New("consumer update refused")
	}
	return nil
}

// A narrowing the bus refuses on a restricted stream removes the consumer, so its wider filters deliver nothing
// more; on an unrestricted stream the refusal is only said (review of PR 338, ADR 0363 §6).
func TestARefusedNarrowingOnARestrictedStreamRemovesTheConsumer(t *testing.T) {
	c := Consumer{Name: "ace_reader", Stream: "MAIL", Filters: mailHeard.Filters[:1]}
	r := &refusingRaiser{refuse: true}
	err := EnsureHeard(r, c, true)
	if err == nil || !strings.Contains(err.Error(), "REMOVED") || !slices.Equal(r.removed, []string{"MAIL/ace_reader"}) {
		t.Fatalf("restricted: %v, removed %v", err, r.removed)
	}
	r = &refusingRaiser{refuse: true}
	if err := EnsureHeard(r, c, false); err == nil || len(r.removed) != 0 {
		t.Fatalf("unrestricted: %v, removed %v", err, r.removed)
	}
	r = &refusingRaiser{}
	if err := EnsureHeard(r, c, true); err != nil || len(r.removed) != 0 {
		t.Fatalf("accepted: %v, removed %v", err, r.removed)
	}
}