mesh-tools/node-tools/internal/meshtest/meshtest.go

// Package meshtest raises what the controller would, for tests against a real bus: the ASSIGNMENTS
// and EVENTS streams, memberships issued by hand, and a fixture's path.
//
// **Every test has a bus of its own** (novox/hq ADR 0237, the controller's internal/testbus): a server
// linked in at the release go.mod pins — the release the mesh runs, held so by a test beside this one —
// started for the one test and gone after it. The packages shared one bus and had to run one at a time:
// each test raised the streams afresh, and the console discovers every runtime that announces itself on
// the bus (ADR 0197), so a runtime from another package's test was, correctly, found. Now the suite runs
// as Go runs it, in parallel and under the race detector. A person may still point a run at a bus of
// their own with MESH_TEST_NATS_EXTERNAL=1 and MESH_TEST_NATS; then the run is theirs to serialise.
package meshtest

import (
	"context"
	"encoding/json"
	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/membership"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-tools/node-tools/internal/bus"
)

// Version is the release of the server the tests run.
const Version = server.VERSION

// URL is the bus of this test alone: started at its first ask, the same one at every ask after — a test
// that dials twice, or hands the address to the code it tests, reaches one bus — and shut down when the
// test ends.
func URL(t *testing.T) string {
	t.Helper()
	if os.Getenv("MESH_TEST_NATS_EXTERNAL") == "1" {
		if url := os.Getenv("MESH_TEST_NATS"); url != "" {
			return url
		}
		t.Fatal("MESH_TEST_NATS_EXTERNAL=1 and MESH_TEST_NATS names no bus")
	}
	if url, ok := buses.Load(t); ok {
		return url.(string)
	}
	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
		NoLog: true, NoSigs: true}
	s, err := server.NewServer(opts)
	if err != nil {
		t.Fatalf("a bus for this test could not be made: %v", err)
	}
	go s.Start()
	if !s.ReadyForConnections(30 * time.Second) {
		s.Shutdown()
		t.Fatal("a bus for this test did not come up within 30s")
	}
	url := s.ClientURL()
	buses.Store(t, url)
	t.Cleanup(func() {
		buses.Delete(t)
		s.Shutdown()
		s.WaitForShutdown()
	})
	return url
}

// buses are the running tests' buses, by test.
var buses sync.Map

// Mesh is the controller's job, done by hand.
type Mesh struct {
	nc *nats.Conn
	js nats.JetStreamContext
}

// New raises the streams afresh.
func New(t *testing.T) *Mesh {
	t.Helper()
	nc, err := nats.Connect(URL(t))
	if err != nil {
		t.Fatal(err)
	}
	js, _ := nc.JetStream()
	_ = js.DeleteStream("EVENTS")
	// The grants part's state memberships, where a runtime reads what is issued it (ADR 0571 clause 10).
	_ = js.DeleteKeyValue(buscompose.GrantsMembershipsBucket)
	if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil {
		t.Fatal(err)
	}
	if _, err := js.AddStream(&nats.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	return &Mesh{nc: nc, js: js}
}

// Issue writes a membership where the grants part keeps it, its state memberships, key `<node>.<module>` (novox/hq ADR
// 0571 clause 7 and 10; mesh-sdk buscompose.GrantsMembershipsBucket), the one place a runtime reads it. It writes with
// mesh-sdk's encoder, the grants part's own, never json of this package's type: a value built here by hand agreed with
// the reader's mistake and hid that the part wrote another shape (issue 774).
func (m *Mesh) Issue(t *testing.T, mem bus.Membership) {
	t.Helper()
	m.putGrants(t, mem, membership.EncodeGrantsMembership(sdkMembership(t, mem)))
}

// IssueInGrantsWrapped writes a membership as the grants part wrote it before mesh-sdk's codec (until 2026-10-30, what
// the live mesh holds until the part writes again): its pass record, the membership inside, published saying what
// became of it ("published", "unchanged", or why it was held back).
func (m *Mesh) IssueInGrantsWrapped(t *testing.T, mem bus.Membership, published string) {
	t.Helper()
	body, _ := json.Marshal(map[string]any{"node": mem.Node, "module": mem.Module, "digest": "sha256:0",
		"membership": sdkMembership(t, mem), "published": published})
	m.putGrants(t, mem, body)
}

func (m *Mesh) putGrants(t *testing.T, mem bus.Membership, body []byte) {
	t.Helper()
	kv, err := m.js.KeyValue(buscompose.GrantsMembershipsBucket)
	if err != nil {
		if kv, err = m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buscompose.GrantsMembershipsBucket, History: 1}); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := kv.Put(buscompose.MembershipKey(mem.Node, mem.Module), body); err != nil {
		t.Fatal(err)
	}
}

// DropGrants deletes the grants part's state memberships, as a bus that has not made it yet holds none; Issue makes it again.
func (m *Mesh) DropGrants(t *testing.T) {
	t.Helper()
	if err := m.js.DeleteKeyValue(buscompose.GrantsMembershipsBucket); err != nil {
		t.Fatal(err)
	}
}

// sdkMembership is a membership of this package's type as mesh-sdk's, which the grants part composes and writes.
func sdkMembership(t *testing.T, mem bus.Membership) membership.Membership {
	t.Helper()
	raw, _ := json.Marshal(mem)
	var out membership.Membership
	if err := json.Unmarshal(raw, &out); err != nil {
		t.Fatal(err)
	}
	return out
}

// ConsoleReads are the states a console's runtime reads to say what the mesh holds (novox/hq ADR 0571 clauses 4, 5):
// placement, the indexer's modules and manifests, and the seat holders.
func ConsoleReads() []bus.StateIssued {
	return []bus.StateIssued{
		{Name: delivery.PlacementRead, Bucket: delivery.PlacementBucket},
		{Name: delivery.IndexerModule + ".modules", Bucket: delivery.ModulesBucket},
		{Name: delivery.IndexerModule + ".manifests", Bucket: delivery.ManifestsBucket},
		{Name: delivery.SeatHoldersRead, Bucket: delivery.SeatHoldersBucket},
	}
}

// Placed writes what the mesh holds as its parts write it, with mesh-sdk's encoders: each module the indexer holds with
// its manifest (tools, the tools it declares; none for a module without), and each node's placement of modules.
func (m *Mesh) Placed(t *testing.T, tools map[string][]string, on map[string][]string) {
	t.Helper()
	kv := func(bucket string) nats.KeyValue {
		_ = m.js.DeleteKeyValue(bucket)
		b, err := m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 1, MaxValueSize: 256 * 1024})
		if err != nil {
			t.Fatal(err)
		}
		return b
	}
	modules, manifests, placement := kv(delivery.ModulesBucket), kv(delivery.ManifestsBucket), kv(delivery.PlacementBucket)
	kv(delivery.SeatHoldersBucket)
	for module, declared := range tools {
		manifest, _ := json.Marshal(map[string]any{"module": module, "tools": declared})
		digest := delivery.ManifestDigest(manifest)
		raw, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: string(manifest)})
		if err != nil {
			t.Fatal(err)
		}
		if _, err := manifests.Put(digest, raw); err != nil {
			t.Fatal(err)
		}
		raw, err = delivery.EncodeModule(delivery.Module{None: "not built here", Manifest: digest, Repository: "novox/" + module,
			Commit: strings.Repeat("a", 40)})
		if err != nil {
			t.Fatal(err)
		}
		if _, err := modules.Put(module, raw); err != nil {
			t.Fatal(err)
		}
	}
	for node, placed := range on {
		p := delivery.Placement{V: delivery.Version, Node: node}
		for _, module := range placed {
			p.Modules = append(p.Modules, delivery.PlacedModule{Module: module})
		}
		// As the seat writes it: its node key and a key per module (research 072 row 37).
		values, err := delivery.PlacementValues(p)
		if err != nil {
			t.Fatal(err)
		}
		for key, raw := range values {
			if _, err := placement.Put(key, raw); err != nil {
				t.Fatal(err)
			}
		}
	}
}

// NextEvent is the subject the next event under a pattern lands on.
func (m *Mesh) NextEvent(t *testing.T, pattern string) <-chan string {
	t.Helper()
	ch := make(chan string, 1)
	sub, err := m.nc.Subscribe(pattern, func(msg *nats.Msg) {
		select {
		case ch <- msg.Subject:
		default:
		}
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	_ = m.nc.Flush()
	return ch
}

// MembershipOf is a membership as the controller issues one on a machine.
func MembershipOf(module, node string, plain bool, seats map[string][]string) bus.Membership {
	own := "mesh.mod." + module
	serves := []bus.Served{{Subject: own + ".tool.{tool}." + node}}
	if plain {
		serves = append(serves, bus.Served{Subject: own + ".tool.{tool}", Queue: "serve." + module})
	}
	var verbs []bus.SeatVerb
	for seat, vs := range seats {
		for _, v := range vs {
			verbs = append(verbs, bus.SeatVerb{Seat: seat, Verb: v, Subject: "mesh.seat." + seat + ".tool." + v + "." + node})
		}
	}
	return bus.Membership{Node: node, Module: module, Serves: serves, Seats: verbs,
		Emits: own + ".event.{event}", Tools: own + ".tool.tools"}
}

// Fixture is a path in node-tools/test/fixtures, which these tests share with the TypeScript ones.
func Fixture(name string) string {
	_, here, _, _ := runtime.Caller(0)
	return filepath.Join(filepath.Dir(here), "..", "..", "test", "fixtures", name)
}

// Until retries while the bus answers "no responders" — something not yet served.
func Until(t *testing.T, try func() error) {
	t.Helper()
	var err error
	for i := 0; i < 50; i++ {
		if err = try(); err == nil {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Fatal(err)
}

// Logs collects what the runtime says.
type Logs struct{ lines []string }

// Logf is a logger that keeps the lines.
func (l *Logs) Logf(format string, args ...any) {
	l.lines = append(l.lines, sprintf(format, args...))
}

// Has says whether a line contains every fragment.
func (l *Logs) Has(fragments ...string) bool {
	for _, line := range l.lines {
		all := true
		for _, f := range fragments {
			all = all && strings.Contains(line, f)
		}
		if all {
			return true
		}
	}
	return false
}

// All is every line.
func (l *Logs) All() string { return strings.Join(l.lines, "\n") }

// Consumer makes a module's durable consumer on a machine as the controller does: pull, on the
// EVENTS stream, filtered to what the module consumes, with a short ack wait so a test sees a
// redelivery in seconds rather than the mesh's minutes.
func (m *Mesh) Consumer(t *testing.T, node, module string, filters []string, ackWait time.Duration) {
	t.Helper()
	cfg := &nats.ConsumerConfig{Durable: node + "_" + module, AckPolicy: nats.AckExplicitPolicy,
		AckWait: ackWait, MaxDeliver: 10, DeliverPolicy: nats.DeliverNewPolicy}
	if len(filters) == 1 {
		cfg.FilterSubject = filters[0]
	} else {
		cfg.FilterSubjects = filters
	}
	if _, err := m.js.AddConsumer("EVENTS", cfg); err != nil {
		t.Fatal(err)
	}
}

// Emit publishes an event as a module would, into the EVENTS stream.
func (m *Mesh) Emit(t *testing.T, subject string, body any) {
	t.Helper()
	data, _ := json.Marshal(body)
	if _, err := m.js.Publish(subject, data); err != nil {
		t.Fatal(err)
	}
}

// Publish says body on a plain subject, as a core publish with no stream behind it.
func (m *Mesh) Publish(t *testing.T, subject string, body any) {
	t.Helper()
	data, _ := json.Marshal(body)
	if err := m.nc.Publish(subject, data); err != nil {
		t.Fatal(err)
	}
	if err := m.nc.Flush(); err != nil {
		t.Fatal(err)
	}
}

// Pending is what a module's consumer still holds: delivered and not acknowledged, and not yet
// delivered.
func (m *Mesh) Pending(t *testing.T, node, module string) (ackPending, notDelivered uint64) {
	t.Helper()
	info, err := m.js.ConsumerInfo("EVENTS", node+"_"+module)
	if err != nil {
		t.Fatal(err)
	}
	return uint64(info.NumAckPending), info.NumPending
}

// BucketKeeping makes a module's state afresh as Bucket does, keeping up to history values a key.
func (m *Mesh) BucketKeeping(t *testing.T, name string, history uint8) nats.KeyValue {
	t.Helper()
	_ = m.js.DeleteKeyValue(name)
	kv, err := m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: name, History: history, MaxValueSize: 256 * 1024})
	if err != nil {
		t.Fatal(err)
	}
	return kv
}

// Bucket makes a module's state afresh as the controller does from the catalogue (novox/hq ADR 0201),
// and answers it for writing what is there before a bundle starts.
func (m *Mesh) Bucket(t *testing.T, name string) nats.KeyValue {
	t.Helper()
	_ = m.js.DeleteKeyValue(name)
	kv, err := m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: name, History: 1, MaxValueSize: 256 * 1024})
	if err != nil {
		t.Fatal(err)
	}
	return kv
}

// BucketKeyTTL makes a module's state afresh as the controller does for one declared `key-ttl` (mesh-sdk buscompose:
// a time to live per key allowed, its end leaving a marker kept a minute; issue 1952).
func (m *Mesh) BucketKeyTTL(t *testing.T, name string) {
	t.Helper()
	_ = m.js.DeleteKeyValue(name)
	js, err := jetstream.New(m.nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 1, MaxValueSize: 256 * 1024,
		LimitMarkerTTL: time.Minute}); err != nil {
		t.Fatal(err)
	}
}

// Log makes a module's log afresh as the controller does from the catalogue (novox/hq ADR 0297): a file
// stream named LOG_<module>_<name> on mesh.log.<module>.<name>.>, limits retention, no age, no cap per
// key, the declared cap, 260 KiB a message (a 256 KiB value and its headers), refusing new entries when full, direct get, no delete or
// purge. It answers the membership entry the controller issues for it.
func (m *Mesh) Log(t *testing.T, module, name string, maxBytes int64) bus.LogIssued {
	t.Helper()
	stream := "LOG_" + module + "_" + name
	subject := "mesh.log." + module + "." + name
	_ = m.js.DeleteStream(stream)
	if _, err := m.js.AddStream(&nats.StreamConfig{Name: stream, Subjects: []string{subject + ".>"},
		Storage: nats.FileStorage, Retention: nats.LimitsPolicy, MaxAge: 0, MaxMsgsPerSubject: -1,
		MaxBytes: maxBytes, MaxMsgSize: 260 * 1024, Discard: nats.DiscardNew, AllowDirect: true,
		DenyDelete: true, DenyPurge: true, Replicas: 1,
		Description: module + "'s log " + name + ", kept as its record (novox/hq ADR 0297)"}); err != nil {
		t.Fatal(err)
	}
	return bus.LogIssued{Name: name, Stream: stream, Subject: subject, Writes: true}
}

// Consumers is how many consumers a stream has.
func (m *Mesh) Consumers(t *testing.T, stream string) int {
	t.Helper()
	info, err := m.js.StreamInfo(stream)
	if err != nil {
		t.Fatal(err)
	}
	return info.State.Consumers
}

// ModuleStream makes a module's own stream afresh as the controller does from the catalogue (novox/hq ADR
// 0363 §1): named in capitals, taking `mesh.stream.<module>.event.>`, a subject space EVENTS does not take.
func (m *Mesh) ModuleStream(t *testing.T, module, name string) {
	t.Helper()
	_ = m.js.DeleteStream(name)
	if _, err := m.js.AddStream(&nats.StreamConfig{Name: name, Subjects: []string{"mesh.stream." + module + ".event.>"},
		Retention: nats.LimitsPolicy, Discard: nats.DiscardOld}); err != nil {
		t.Fatal(err)
	}
}

// ConsumerOn makes a module's durable consumer on a machine on any stream, as Consumer does on EVENTS.
func (m *Mesh) ConsumerOn(t *testing.T, stream, node, module string, filters []string, ackWait time.Duration) {
	t.Helper()
	cfg := &nats.ConsumerConfig{Durable: node + "_" + module, AckPolicy: nats.AckExplicitPolicy,
		AckWait: ackWait, MaxDeliver: 10, DeliverPolicy: nats.DeliverNewPolicy}
	if len(filters) == 1 {
		cfg.FilterSubject = filters[0]
	} else {
		cfg.FilterSubjects = filters
	}
	if _, err := m.js.AddConsumer(stream, cfg); err != nil {
		t.Fatal(err)
	}
}

// PendingOn is what a module's consumer on a stream still holds, as Pending is on EVENTS.
func (m *Mesh) PendingOn(t *testing.T, stream, node, module string) (ackPending, notDelivered uint64) {
	t.Helper()
	info, err := m.js.ConsumerInfo(stream, node+"_"+module)
	if err != nil {
		t.Fatal(err)
	}
	return uint64(info.NumAckPending), info.NumPending
}

// Messages is how many messages a stream holds.
func (m *Mesh) Messages(t *testing.T, stream string) uint64 {
	t.Helper()
	info, err := m.js.StreamInfo(stream)
	if err != nil {
		t.Fatal(err)
	}
	return info.State.Msgs
}

// RemoveConsumer deletes a module's consumer on a machine on a stream, as the controller does when a grant
// it cannot narrow is withdrawn (novox/hq ADR 0363 §6).
func (m *Mesh) RemoveConsumer(t *testing.T, stream, node, module string) {
	t.Helper()
	if err := m.js.DeleteConsumer(stream, node+"_"+module); err != nil {
		t.Fatal(err)
	}
}