mesh-delivery/modules/mesh-grants/cmd/mesh-grants/integration_test.go

//go:build integration

package main

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

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

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/membership"
	"git.novox.be/novox/mesh-sdk/go/process"
	"git.novox.be/novox/mesh-sdk/go/seal"
	"git.novox.be/novox/mesh-sdk/go/views"

	"mesh-grants/internal/grants"
)

// TestTheGrantsPartWhole starts the program the way the node does (its entry, run, with its environment and a broker
// file) against an embedded bus that holds no controller at all: no lease bucket, no controller placed (novox/hq ADR 0590,
// issue 1126). Given a module placed with an event it consumes, it publishes the module's membership in its own state and
// on no subject, makes the module's durable consumer on EVENTS, and writes the sealed list's key for the bus node to follow
// alone; and it creates no lease, nor any state of a lease write or a take-over.
func TestTheGrantsPartWhole(t *testing.T) {
	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "grants")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(dir) })
	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: filepath.Join(dir, "js"), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go srv.Start()
	if !srv.ReadyForConnections(10 * time.Second) {
		t.Fatal("the bus did not come up")
	}
	t.Cleanup(srv.Shutdown)
	nc, err := nats.Connect(srv.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, _ := jetstream.New(nc)
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	kvs := map[string]jetstream.KeyValue{}
	for _, state := range append(append([]string(nil), grants.Reads...), grants.ViewReads...) {
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucketOf(state), History: 5})
		if err != nil {
			t.Fatal(err)
		}
		kvs[state] = kv
	}
	for _, state := range grants.States {
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: grants.Module + "_" + state, History: 5})
		if err != nil {
			t.Fatal(err)
		}
		kvs[grants.Module+"."+state] = kv
	}
	for _, c := range []jetstream.StreamConfig{
		{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}},
		{Name: "CONDITIONS", Subjects: []string{"mesh.seat.operator-channel.accept.condition.>"}},
	} {
		if _, err := js.CreateStream(ctx, c); err != nil {
			t.Fatal(err)
		}
	}
	put := func(state, key string, v any) {
		t.Helper()
		raw, err := json.Marshal(v)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := kvs[state].Put(ctx, key, raw); err != nil {
			t.Fatal(err)
		}
	}
	manifest := func(module, raw string) {
		digest := delivery.ManifestDigest([]byte(raw))
		put(grants.ReadManifests, digest, delivery.Manifest{V: delivery.Version, Module: module, JSON: raw})
		put(grants.ReadModules, module, delivery.Module{V: delivery.Version, FP2: "src2:" + strings.Repeat("ab", 32), Manifest: digest,
			Repository: "novox/x", Commit: "c"})
		verdict, _ := delivery.EncodeManifestVerdict(delivery.ManifestVerdict{Module: module, State: delivery.VerdictAccepted, Judge: "j", At: time.Now()})
		put(grants.ReadVerdicts, digest, json.RawMessage(verdict))
	}
	manifest("nats", `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"},`+
		`"emits":["created","done","dropped","recreated"],"streams":[{"name":"broker-events","events":["created","dropped","recreated"],"max-age-days":30,"max-mib":256}]}`)
	manifest("mod-c", `{"module":"mod-c","version":"1","own-secrets":{"broker":"/var/lib/c/broker"},"consumes":["nats.done"]}`)
	// The part itself, placed as on the mesh, so its own consumer on EVENTS hears the seat mesh-broker's events (DLV-G1).
	own, err := os.ReadFile("../../module.json")
	if err != nil {
		t.Fatal(err)
	}
	manifest(grants.Module, string(own))
	pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "a", Modules: []delivery.PlacedModule{{Module: "nats"}, {Module: "mod-c"},
		{Module: grants.Module}}})
	if _, err := kvs[grants.ReadPlacement].Put(ctx, "placement.a", pl); err != nil {
		t.Fatal(err)
	}
	busKey, _ := seal.GenerateKeyPair()
	put(grants.ReadHave, "a.node", map[string]any{"v": 1})
	put(grants.ReadHave, "a.nats", delivery.Have{V: delivery.Version, Phase: delivery.PhaseRunning, Health: delivery.HealthOK,
		UserList: &delivery.UserListWritten{Source: delivery.GrantsUserListSource, Revision: 1, Digest: "sha256:aa", Users: 1,
			Key: busKey.PublicKey, Follows: delivery.FollowGrants}})

	// The bus module's read tools, as nats-tools answers them: the streams with their consumers as the bus holds them now.
	answer := func(m *nats.Msg, result any) {
		raw, _ := json.Marshal(map[string]any{"result": result, "node": "a"})
		_ = m.Respond(raw)
	}
	sub, err := nc.Subscribe("mesh.mod.nats.>", func(m *nats.Msg) {
		switch {
		case strings.Contains(m.Subject, ".nats_users"):
			answer(m, map[string]any{"users": []any{}})
		case strings.Contains(m.Subject, ".nats_streams"):
			var args struct{ Stream string }
			_ = json.Unmarshal(m.Data, &args)
			detail := []any{}
			if st, err := js.Stream(ctx, args.Stream); err == nil {
				consumers := []any{}
				names := st.ConsumerNames(ctx)
				for name := range names.Name() {
					c, _ := st.Consumer(ctx, name)
					cfg := c.CachedInfo().Config
					consumers = append(consumers, map[string]any{"name": name, "filters": cfg.FilterSubjects, "max_deliver": cfg.MaxDeliver})
				}
				detail = append(detail, map[string]any{"name": args.Stream, "consumer_detail": consumers})
			}
			answer(m, map[string]any{"detail": detail})
		default:
			answer(m, map[string]any{})
		}
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = sub.Unsubscribe() })
	// DLV-F4r, DLV-F5r (ADR 0615 §4, §6, ADR 0621 §3): the owners answer their first-view tools at once from what they
	// hold, with its revision, as the indexer's and the assigner's runtimes of their own do; the part never reads their
	// buckets. Started before its owners answer (answering false: each ask refused, as an owner not yet connected), the
	// part waits, says so, and composes nothing it would read from them.
	var firstViewAsks, refusedAsks atomic.Int32
	var answering atomic.Bool
	for _, owner := range []string{grants.IndexerModule, grants.AssignerModule} {
		owner := owner
		osub, err := nc.Subscribe("mesh.mod."+owner+".>", func(m *nats.Msg) {
			for _, fv := range grants.FirstViews {
				if fv.Owner != owner || !strings.Contains(m.Subject, "."+fv.State) {
					continue
				}
				if !answering.Load() {
					refusedAsks.Add(1)
					raw, _ := json.Marshal(map[string]any{"error": "not connected yet", "node": "a"})
					_ = m.Respond(raw)
					return
				}
				firstViewAsks.Add(1)
				values, rev := map[string]json.RawMessage{}, uint64(0)
				if keys, err := kvs[fv.Read].ListKeys(ctx); err == nil {
					for k := range keys.Keys() {
						if e, err := kvs[fv.Read].Get(ctx, k); err == nil {
							values[k] = e.Value()
							rev = max(rev, e.Revision())
						}
					}
				}
				answer(m, views.Answer{V: views.Version, Owner: owner, State: fv.State, Revision: rev,
					Created: time.Date(2026, 11, 4, 0, 0, 0, 0, time.UTC), Values: values, At: time.Now()})
				return
			}
		})
		if err != nil {
			t.Fatal(err)
		}
		t.Cleanup(func() { _ = osub.Unsubscribe() })
	}
	// D-30 (novox/hq ADR 0636 §1): the seat mesh-catalog's holder answers the seat list with no trunk seat named, so the reader
	// has heard the seat and holds the bootstrapped definitions, and every module is composed. (The unit case holds a module
	// while the list is unread.)
	var seatAnswering atomic.Bool
	seatAnswering.Store(true)
	for _, subject := range []string{"mesh.seat.mesh-catalog.call.seat-definitions", "mesh.seat.mesh-catalog.call.seat-definitions.>"} {
		if _, err := nc.Subscribe(subject, func(m *nats.Msg) {
			if !seatAnswering.Load() {
				_ = m.Respond([]byte(`{"error":"not connected yet","node":"a"}`))
				return
			}
			answer(m, map[string]any{"revision": 1, "seats": []any{}})
		}); err != nil {
			t.Fatal(err)
		}
	}
	// A membership is never said on its subject any more (issue 1126).
	var said atomic.Int32
	if _, err := nc.Subscribe("mesh.assignment.>", func(*nats.Msg) { said.Add(1) }); err != nil {
		t.Fatal(err)
	}

	broker := filepath.Join(dir, "broker")
	cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "a", Module: grants.Module, User: "a." + grants.Module, Password: "pw"})
	if err := os.WriteFile(broker, cred, 0o600); err != nil {
		t.Fatal(err)
	}
	env := map[string]string{"MESH_BROKER_FILE": broker}
	runCtx, stop := context.WithCancel(ctx)
	done := make(chan error, 1)
	logged := &lockedLog{}
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, logged) }()
	t.Cleanup(func() { stop(); <-done })

	until := func(what string, ok func() bool) {
		t.Helper()
		for deadline := time.Now().Add(30 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
			if ok() {
				return
			}
		}
		t.Fatalf("never saw: %s", what)
	}
	until("every first view asked and refused once", func() bool { return refusedAsks.Load() >= int32(len(grants.FirstViews)) })
	until("the part saying it waits on the indexer and the assigner", func() bool {
		return logged.has("mesh-delivery-indexer.manifests") && logged.has("mesh-delivery-assigner.placement") && logged.has("is not answered")
	})
	if _, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c"); err == nil {
		t.Error("a membership composed before the owners answered its first views")
	}
	answering.Store(true)
	until("a.mod-c's membership in the part's own state memberships", func() bool {
		e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c")
		if err != nil {
			return false
		}
		m, err := membership.DecodeGrantsMembership(e.Value())
		return err == nil && m.Module == "mod-c"
	})
	until("mod-c's durable consumer on EVENTS, made by the part", func() bool {
		_, err := js.Consumer(ctx, "EVENTS", "a_mod-c")
		return err == nil
	})
	until("the sealed list's key, followed by the bus node alone", func() bool {
		e, err := kvs[grants.Module+"."+grants.StateUserList].Get(ctx, grants.KeySealed)
		if err != nil {
			return false
		}
		g, err := delivery.DecodeGrantsUserList(e.Value())
		return err == nil && g.Follow == delivery.FollowGrants
	})
	if n := said.Load(); n != 0 {
		t.Errorf("%d memberships were said on their subject", n)
	}
	if firstViewAsks.Load() < int32(len(grants.FirstViews)) {
		t.Errorf("the part composed with %d first-view asks, want one of each of the %d", firstViewAsks.Load(), len(grants.FirstViews))
	}
	// DLV-G1, 1966/AC34 (novox/hq ADR 0595 §1, §7): the bus module's created, said on the seat mesh-broker's subject on
	// EVENTS like every seat's event (the lead's ruling, registry milestone 13:0x), reaches the part on its durable consumer
	// there, and the binding it carries is in the module's membership; dropped takes it out.
	until("the part's durable consumer on EVENTS hearing mesh-broker's created", func() bool {
		cons, err := js.Consumer(ctx, "EVENTS", "a_"+grants.Module)
		if err != nil {
			return false
		}
		info, err := cons.Info(ctx)
		return err == nil && slices.Contains(info.Config.FilterSubjects, "mesh.seat.mesh-broker.event.created")
	})
	bindingOf := func() []membership.BusBinding {
		e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c")
		if err != nil {
			return nil
		}
		m, _ := membership.DecodeGrantsMembership(e.Value())
		return m.Bindings
	}
	sayBroker := func(event string, bind membership.BusBinding) {
		body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-c", Local: "kept", Storage: "key-value", Template: "state", Binding: bind})
		if _, err := js.Publish(ctx, "mesh.seat.mesh-broker.event."+event, body); err != nil {
			t.Fatal(err)
		}
	}
	sayBroker("created", membership.BusBinding{Local: "kept", Storage: "key-value", Template: "state", Bucket: "mod-c_kept"})
	until("mod-c's binding in its membership after created", func() bool {
		b := bindingOf()
		return len(b) == 1 && b[0].Bucket == "mod-c_kept"
	})
	sayBroker("dropped", membership.BusBinding{Local: "kept"})
	until("mod-c's binding gone after dropped", func() bool { return len(bindingOf()) == 0 })
	for _, bucket := range []string{"mesh-controller_lease", grants.Module + "_lease", grants.Module + "_takeover"} {
		if _, err := js.KeyValue(ctx, bucket); !errors.Is(err, jetstream.ErrBucketNotFound) {
			t.Errorf("the bucket %s: %v, want none", bucket, err)
		}
	}
	// And what the part composes holds on a server that loads it (one integration test per module, the test budget).
	t.Run("the composed grants hold on a server that loads them", theBusNodeFollowsTheSealedListUnderTheComposedGrants)
}

// lockedLog is the part's log, read by the test while the part writes it.
type lockedLog struct {
	mu  sync.Mutex
	buf strings.Builder
}

func (l *lockedLog) Write(p []byte) (int, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.buf.Write(p)
}

func (l *lockedLog) has(s string) bool {
	l.mu.Lock()
	defer l.mu.Unlock()
	return strings.Contains(l.buf.String(), s)
}