mesh-controller/cmd/mesh-controller/busobjects_test.go

//go:build integration

package main

import (
	"errors"
	"strings"
	"testing"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"github.com/novox/mesh-controller/internal/broker"
	"github.com/novox/mesh-controller/internal/inventory"
)

// novox/hq issue 208: the bus's objects a declaration implies are asserted whenever one is sent, not
// only when the controller starts.

// aCarriedConsumer is the runtime on laptop and, carried by it, a module that consumes an event: the
// shape of messenger on 2026-10-06, assigned after the controller started.
func aCarriedConsumer(t *testing.T, open *stores) broker.Consumer {
	t.Helper()
	register(t, open, catalogue.Manifest{Module: catalogue.RuntimeModule, Version: "1",
		OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/var/lib/mesh/node-tools/broker"}}})
	register(t, open, catalogue.Manifest{Module: "messenger", Version: "1",
		Consumes: []string{"billing.order.placed"}})
	for _, m := range []string{catalogue.RuntimeModule, "messenger"} {
		if _, err := open.inventory.Assign(t.Context(), "laptop", m); err != nil {
			t.Fatal(err)
		}
	}
	consumers, err := moduleConsumers(t.Context(), open.inventory)
	if err != nil {
		t.Fatal(err)
	}
	for _, c := range consumers {
		if c.Module == "messenger" && c.Node == "laptop" {
			return c.Consumer
		}
	}
	t.Fatalf("messenger on laptop is derived no consumer: %+v", consumers)
	return broker.Consumer{}
}

// aLateHolder is a module holding a build seat (node-build-go) on anchor, assigned after the controller started, and
// the worker its seat's queue should have for it: a build seat is pulled by kind, so the worker is the one for the
// seat's first accepted kind (the build agent's seat, which had one worker for all, is gone with the builder).
func aLateHolder(t *testing.T, open *stores) broker.Consumer {
	t.Helper()
	// With an account to read, as a holder has: a module with none is no bus user, so its seat's workers are not derived.
	register(t, open, catalogue.Manifest{Module: "late-builder", Version: "1",
		OwnSecrets: catalogue.OwnSecrets{"broker": {Path: "/var/lib/mesh/late-builder/broker"}},
		Claims:     []catalogue.Claim{{Name: "node-build-go", Scope: catalogue.ScopeNode}}})
	if _, err := open.inventory.Assign(t.Context(), "anchor", "late-builder"); err != nil {
		t.Fatal(err)
	}
	_, workers, err := seatTrafficObjects(t.Context(), open.inventory)
	if err != nil {
		t.Fatal(err)
	}
	for _, s := range inventory.MeshSeats() {
		if s.Name != "node-build-go" || len(s.Accepts) == 0 {
			continue
		}
		for _, w := range workers {
			if w.Name == broker.WorkerName(s.Name, s.Accepts[0]) {
				return w
			}
		}
	}
	t.Fatal("the build seat's holder is derived no worker")
	return broker.Consumer{}
}

// asserted says whether a recording holds a consumer by stream and name.
func asserted(rec *recordingRaiser, want broker.Consumer) bool {
	for _, c := range rec.consumers {
		if c.Stream == want.Stream && c.Name == want.Name {
			return true
		}
	}
	return false
}

// What a send asserts includes what was assigned after the start's assertion: a carried module's
// consumer and a late holder's worker. The derivation is the start's own, so this holds without a bus.
func TestASendAssertsWhatWasAssignedAfterStart(t *testing.T) {
	open := aMesh(t)
	var atStart recordingRaiser
	if _, err := assertBusObjects(t.Context(), open.inventory, &atStart); err != nil {
		t.Fatal(err)
	}
	consumer := aCarriedConsumer(t, open)
	worker := aLateHolder(t, open)
	if asserted(&atStart, consumer) || asserted(&atStart, worker) {
		t.Fatal("the start asserted what was not yet assigned; the test proves nothing")
	}
	var onSend recordingRaiser
	if err := assertOnSend(t.Context(), open.inventory, &onSend, ""); err != nil {
		t.Fatal(err)
	}
	if !asserted(&onSend, consumer) {
		t.Fatalf("messenger's consumer %s on %s is not asserted by the send: %+v", consumer.Name, consumer.Stream, onSend.consumers)
	}
	if !asserted(&onSend, worker) {
		t.Fatalf("the late holder's worker %s on %s is not asserted by the send: %+v", worker.Name, worker.Stream, onSend.consumers)
	}
}

// failingRaiser refuses one consumer by name and records everything it was asked.
type failingRaiser struct {
	recordingRaiser
	refuse string
}

func (f *failingRaiser) EnsureConsumer(c broker.Consumer) error {
	f.consumers = append(f.consumers, c)
	if c.Name == f.refuse {
		return errors.New("nats: API error: code=503 description=insufficient resources")
	}
	return nil
}

// A send whose objects cannot be asserted says so in its output and raises a condition — and the next
// send that asserts them clears it. The consumers after the refused one are still asked for.
func TestASendThatCannotAssertTheBusSaysSoAndRaisesACondition(t *testing.T) {
	open := aMesh(t)
	consumer := aCarriedConsumer(t, open)
	worker := aLateHolder(t, open)
	failing := &failingRaiser{refuse: consumer.Name}

	var sendErr error
	out := stdoutOf(t, func() error {
		sendErr = assertOnSend(t.Context(), open.inventory, failing, "  ")
		return nil
	})
	if sendErr == nil || !strings.Contains(sendErr.Error(), "how messenger on laptop hears what it consumes") {
		t.Fatalf("the failure is not answered: %v", sendErr)
	}
	if !strings.Contains(out, "THE BUS DOES NOT HOLD WHAT THIS SEND IMPLIES") ||
		!strings.Contains(out, "insufficient resources") || !strings.Contains(out, "bus.objects.unasserted") {
		t.Fatalf("the failure is not said in the send's output:\n%s", out)
	}
	if !asserted(&failing.recordingRaiser, worker) {
		t.Fatal("the seat's worker was not asked for")
	}
	c, open1, err := conditionsFrom.Get(t.Context(), "bus.objects.unasserted")
	if err != nil || !open1 {
		t.Fatalf("no condition raised: %v", err)
	}
	if c.Kind != kindBusObjectsUnasserted || c.Source != sourceBusObjects ||
		len(c.Evidence) == 0 || !strings.Contains(c.Evidence[0].Said, "insufficient resources") {
		t.Fatalf("the condition does not say what failed: %+v", c)
	}

	// The next send asserts them, and that observation clears it.
	if err := assertOnSend(t.Context(), open.inventory, &recordingRaiser{}, ""); err != nil {
		t.Fatal(err)
	}
	if _, stillOpen, err := conditionsFrom.Get(t.Context(), "bus.objects.unasserted"); err != nil || stillOpen {
		t.Fatalf("a send that asserted everything left the condition open: %v", err)
	}
}