mesh-controller/internal/broker/agreement_catalogue_test.go

package broker

import (
	"encoding/json"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"testing"

	"github.com/novox/mesh-controller/internal/catalogue"
)

// **Do the catalogue's emitters and consumers agree?**
//
// This is the check whose absence let issue 127 stand: every manifest was individually well-formed,
// every derivation individually correct, and no cross-module subscription in the mesh matched
// anything. A subscription that matches nothing is not an error — it is silence — so nothing
// anywhere reported it.
//
// It compares what one manifest asks to hear against what another says it emits. It cannot demand
// that every consumed event have a live emitter, because a module lives in its own repository and
// may be registered long before the one whose events it wants. Where the emitter *is* here, it must
// emit what the consumer asked for.
func TestTheCataloguesEmittersAndConsumersAgree(t *testing.T) {
	emitters, consumers, seats := theCataloguesEvents(t)

	if bad := Disagreements(emitters, consumers, seats); len(bad) > 0 {
		t.Fatalf("%d subscription(s) in the catalogue would match nothing:\n  %s",
			len(bad), strings.Join(bad, "\n  "))
	}
}

// And the check itself catches the thing it exists for, so it cannot pass by doing nothing.
func TestTheAgreementCheckCatchesASubscriptionThatMatchesNothing(t *testing.T) {
	bad := Disagreements(
		[]AnEmitter{{Module: "builder", Emits: []string{"built"}}},
		[]AConsumer{{Module: "mesh-catalog", Consumes: []string{"builder.finished"}}},
		nil)
	if len(bad) != 1 {
		t.Fatalf("a consumer asking for an event its emitter does not emit was not caught: %v", bad)
	}
	if !strings.Contains(bad[0], "builder.finished") || !strings.Contains(bad[0], "built") {
		t.Fatalf("the report names neither what was asked for nor what is emitted: %s", bad[0])
	}

	// A module that is not here is not a disagreement: it may be registered later.
	if bad := Disagreements(nil,
		[]AConsumer{{Module: "plex", Consumes: []string{"sonarr.download.completed"}}}, nil); len(bad) != 0 {
		t.Fatalf("a consumer whose emitter is not installed was reported: %v", bad)
	}

	// A wildcard over emitters is deliberate and names no particular event to check.
	if bad := Disagreements([]AnEmitter{{Module: "sonarr", Emits: []string{"download.completed"}}},
		[]AConsumer{{Module: "plex", Consumes: []string{"*.download.completed"}}}, nil); len(bad) != 0 {
		t.Fatalf("a wildcard over emitters was reported: %v", bad)
	}

	// A consumer of a role's event whose role does not emit it is caught, which is what stops the
	// catalogue check above from passing by knowing nothing about roles.
	if bad := Disagreements(nil,
		[]AConsumer{{Module: "mesh-catalog", Consumes: []string{"mesh-build-machine.finished"}}},
		[]DeclaredSeat{{Name: "mesh-build-machine", Emits: []string{"built"}}}); len(bad) != 1 {
		t.Fatalf("a consumer of a role event the role does not emit was not caught: %v", bad)
	}

	// An event published under a seat's name is real even though no module declares it as its own.
	if bad := Disagreements(nil,
		[]AConsumer{{Module: "watcher", Consumes: []string{"mesh-artifact-store.image.pushed"}}},
		[]DeclaredSeat{{Name: "mesh-artifact-store", Emits: []string{"image.pushed"}}}); len(bad) != 0 {
		t.Fatalf("an event a seat emits was reported as matching nothing: %v", bad)
	}
}

func theCataloguesEvents(t *testing.T) ([]AnEmitter, []AConsumer, []DeclaredSeat) {
	t.Helper()
	root := filepath.Join("..", "..", "..", "mesh-catalog", "modules")
	entries, err := os.ReadDir(root)
	if err != nil {
		t.Skipf("catalogue sibling not present: %v", err)
	}
	var emitters []AnEmitter
	var consumers []AConsumer
	// The mesh's own roles, which emit under the seat's name rather than any module's (novox/hq
	// ADR 0121). Without these the check skips every consumer of a role's event as "the emitter is
	// not installed" — which is how it passed vacuously the first time one existed.
	var seats []DeclaredSeat
	for _, own := range catalogue.SeatsWithAProtocol() {
		seats = append(seats, DeclaredSeat{Name: own.Name, Accepts: own.Accepts, Emits: own.Emits})
	}
	for _, e := range entries {
		if !e.IsDir() {
			continue
		}
		raw, err := os.ReadFile(filepath.Join(root, e.Name(), "module.json"))
		if err != nil {
			continue
		}
		var m struct {
			Module   string   `json:"module"`
			Emits    []string `json:"emits"`
			Consumes []string `json:"consumes"`
			Seats    []struct {
				Name  string   `json:"name"`
				Emits []string `json:"emits"`
			} `json:"seats"`
		}
		if err := json.Unmarshal(raw, &m); err != nil {
			t.Fatalf("%s: %v", e.Name(), err)
		}
		if len(m.Emits) > 0 {
			emitters = append(emitters, AnEmitter{Module: m.Module, Emits: m.Emits})
		}
		if len(m.Consumes) > 0 {
			consumers = append(consumers, AConsumer{Module: m.Module, Consumes: m.Consumes})
		}
		for _, s := range m.Seats {
			seats = append(seats, DeclaredSeat{Name: s.Name, Emits: s.Emits})
		}
	}
	if len(emitters) == 0 {
		t.Skip("no manifests found beside this checkout")
	}
	return emitters, consumers, seats
}

// **Do the derived subjects meet, not just the names?**
//
// The check above compares what a consumer asks for against what an emitter says it emits, by name. It
// passed while the catalogue's subscription pointed at `mesh.mod.mesh-build-machine.event.built` — a
// module namespace for a role's event, which no emitter owns. The names agreed; the subjects did not,
// and the graph stayed empty.
//
// So this compares the thing that actually has to match: the subject a consumer subscribes against the
// subject an emitter publishes. It is the last place the two halves can be held together, because
// after this the server is the only thing that knows and it says nothing — a subscription that matches
// nothing is silence.
func TestTheCataloguesDerivedSubjectsMeet(t *testing.T) {
	emitters, consumers, seats := theCataloguesEvents(t)

	// Every subject something publishes: a module's own events, and the events of every role.
	published := map[string]bool{}
	for _, e := range emitters {
		for _, name := range e.Emits {
			published["mesh.mod."+e.Module+".event."+name] = true
		}
	}
	for _, s := range seats {
		for _, name := range s.Emits {
			published["mesh.seat."+s.Name+".event."+name] = true
		}
	}

	byName := map[string]DeclaredSeat{}
	for _, s := range seats {
		byName[s.Name] = s
	}

	var lonely []string
	for _, c := range consumers {
		principal := Principal{Kind: KindModule, Node: "one", Module: c.Module, PasswordHash: "x"}
		for _, want := range c.Consumes {
			emitter, event, named := strings.Cut(want, ".")
			if named {
				if s, isASeat := byName[emitter]; isASeat {
					principal.Watches = append(principal.Watches,
						Seat{Name: s.Name, Emits: []string{event}})
					continue
				}
			}
			principal.Consumes = append(principal.Consumes, want)
		}
		perms, err := PermissionsFor(principal)
		if err != nil {
			t.Fatalf("%s: %v", c.Module, err)
		}
		for _, subject := range perms.Subscribe {
			if !strings.Contains(subject, ".event.") {
				continue
			}
			if reaches(subject, published) {
				continue
			}
			// A wildcard over emitters reaches whatever arrives later, and an emitter that is not
			// installed is ordinary — both are already excused by the check above, so only a subject
			// that can never match anything gets here.
			if strings.Contains(subject, "*") || strings.Contains(subject, ">") {
				continue
			}
			lonely = append(lonely, c.Module+" subscribes "+subject+", which nothing publishes")
		}
	}
	if len(lonely) > 0 {
		sort.Strings(lonely)
		t.Fatalf("%d subscription(s) derive to a subject no emitter owns:\n  %s",
			len(lonely), strings.Join(lonely, "\n  "))
	}
}

// And it catches the thing it exists for: a role's event read as a module's.
func TestTheDerivedSubjectCheckCatchesARolesEventReadAsAModules(t *testing.T) {
	published := map[string]bool{"mesh.seat.mesh-build-machine.event.built": true}
	// What the derivation produced before a consumed seat name was resolved as one.
	if reaches("mesh.mod.mesh-build-machine.event.built", published) {
		t.Fatal("a module namespace was treated as reaching a role's event, which is the bug")
	}
	// And the corrected one does reach it.
	if !reaches("mesh.seat.mesh-build-machine.event.built", published) {
		t.Fatal("the role's own subject does not reach the role's event")
	}
}

// reaches says whether a subscribed subject admits any published one.
func reaches(subject string, published map[string]bool) bool {
	for p := range published {
		if admitsSubject(strings.Split(subject, "."), strings.Split(p, ".")) {
			return true
		}
	}
	return false
}

func admitsSubject(pattern, subject []string) bool {
	for i, token := range pattern {
		if token == ">" {
			return i < len(subject)
		}
		if i >= len(subject) {
			return false
		}
		if token != "*" && token != subject[i] {
			return false
		}
	}
	return len(pattern) == len(subject)
}