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

//go:build integration

package main

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"os"
	"path/filepath"
	"strings"
	"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"

	"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 grants.Reads {
		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"})
	}
	manifest("nats", `{"module":"nats","version":"1","claims":[{"name":"mesh-broker"}],"own-secrets":{"broker":"/var/lib/nats/broker"}}`)
	manifest("mod-c", `{"module":"mod-c","version":"1","own-secrets":{"broker":"/var/lib/c/broker"},"consumes":["nats.done"]}`)
	pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "a", Modules: []delivery.PlacedModule{{Module: "nats"}, {Module: "mod-c"}}})
	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() })
	// 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)
	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
	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("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)
	}
	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)
		}
	}
}