mesh-controller/cmd/mesh-controller/composed_writer_test.go

package main

import (
	"compress/gzip"
	"context"
	"crypto/ed25519"
	"crypto/rand"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"sort"
	"strings"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go/jetstream"
)

// fakeComposedKV is a key-value bucket in memory: revisions, a creation time per write, history, compare-and-set.
type fakeComposedKV struct {
	name    string
	rev     uint64
	clock   time.Time
	history map[string][]fakeComposedEntry
	writes  int
	deletes int
}

type fakeComposedEntry struct {
	bucket, key string
	value       []byte
	rev         uint64
	at          time.Time
	op          jetstream.KeyValueOp
}

func (e fakeComposedEntry) Bucket() string                  { return e.bucket }
func (e fakeComposedEntry) Key() string                     { return e.key }
func (e fakeComposedEntry) Value() []byte                   { return e.value }
func (e fakeComposedEntry) Revision() uint64                { return e.rev }
func (e fakeComposedEntry) Created() time.Time              { return e.at }
func (e fakeComposedEntry) Delta() uint64                   { return 0 }
func (e fakeComposedEntry) Operation() jetstream.KeyValueOp { return e.op }

func newFakeComposedKV(name string) *fakeComposedKV {
	return &fakeComposedKV{name: name, clock: time.Date(2026, 10, 24, 17, 0, 0, 0, time.UTC), history: map[string][]fakeComposedEntry{}}
}

func (f *fakeComposedKV) last(key string) (fakeComposedEntry, bool) {
	h := f.history[key]
	if len(h) == 0 || h[len(h)-1].op != jetstream.KeyValuePut {
		return fakeComposedEntry{}, false
	}
	return h[len(h)-1], true
}

func (f *fakeComposedKV) put(key string, value []byte, op jetstream.KeyValueOp) uint64 {
	f.rev++
	f.clock = f.clock.Add(time.Second)
	f.history[key] = append(f.history[key], fakeComposedEntry{f.name, key, value, f.rev, f.clock, op})
	return f.rev
}

func (f *fakeComposedKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) {
	e, ok := f.last(key)
	if !ok {
		return nil, jetstream.ErrKeyNotFound
	}
	return e, nil
}

func (f *fakeComposedKV) Create(_ context.Context, key string, value []byte, _ ...jetstream.KVCreateOpt) (uint64, error) {
	if _, ok := f.last(key); ok {
		return 0, jetstream.ErrKeyExists
	}
	f.writes++
	return f.put(key, value, jetstream.KeyValuePut), nil
}

func (f *fakeComposedKV) Update(_ context.Context, key string, value []byte, revision uint64) (uint64, error) {
	e, ok := f.last(key)
	if !ok || e.rev != revision {
		return 0, fmt.Errorf("wrong last sequence")
	}
	f.writes++
	return f.put(key, value, jetstream.KeyValuePut), nil
}

func (f *fakeComposedKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
	if _, ok := f.last(key); !ok {
		return jetstream.ErrKeyNotFound
	}
	f.deletes++
	f.put(key, nil, jetstream.KeyValueDelete)
	return nil
}

func (f *fakeComposedKV) History(_ context.Context, key string, _ ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error) {
	h := f.history[key]
	if len(h) == 0 {
		return nil, jetstream.ErrKeyNotFound
	}
	out := make([]jetstream.KeyValueEntry, len(h))
	for i, e := range h {
		out[i] = e
	}
	return out, nil
}

type fakeLister struct{ keys chan string }

func (l fakeLister) Keys() <-chan string { return l.keys }
func (l fakeLister) Stop() error         { return nil }

func (f *fakeComposedKV) ListKeys(_ context.Context, _ ...jetstream.WatchOpt) (jetstream.KeyLister, error) {
	var keys []string
	for k := range f.history {
		if _, ok := f.last(k); ok {
			keys = append(keys, k)
		}
	}
	sort.Strings(keys)
	ch := make(chan string, len(keys))
	for _, k := range keys {
		ch <- k
	}
	close(ch)
	return fakeLister{ch}, nil
}

// fakeObjects is a node's object bucket in memory.
type fakeObjects map[string][]byte

func (o fakeObjects) Has(_ context.Context, name string) (bool, bool, error) {
	_, ok := o[name]
	return ok, false, nil
}
func (o fakeObjects) Put(_ context.Context, name string, content []byte) error {
	o[name] = content
	return nil
}

// aPlace is a node's buckets in memory, signed with a key of the test's, the want and the have given.
func aPlace(t *testing.T, node string) (composedPlace, ed25519.PublicKey, map[string]string, map[string]sdkdelivery.Have) {
	t.Helper()
	pub, priv, err := ed25519.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	wants, haves := map[string]string{}, map[string]sdkdelivery.Have{}
	at := composedPlace{node: node, bucket: newFakeComposedKV("composed-" + node), digests: newFakeComposedKV("mesh-controller_composed"),
		objects: fakeObjects{},
		wanted:  func(_ context.Context, module string) (string, error) { return wants[module], nil },
		ran: func(_ context.Context, module string) (sdkdelivery.Have, bool, error) {
			h, ok := haves[module]
			return h, ok, nil
		},
		policy: func(string) (string, string, string) { return catalogue.PolicyRoll, catalogue.FromDefault, "" },
		sign:   func(c sdkdelivery.Composed) (sdkdelivery.Composed, error) { return sdkdelivery.SignComposed(c, priv) },
	}
	return at, pub, wants, haves
}

// capturedBody is a node's declaration as the controller's `plan` verb answered it on 2026-10-24 at 16:19, the body a
// send of declarationFor carries (testdata/declaration-<node>.json.gz, copied from mesh-sdk's; the resources carrying
// a sealed value or a secret taken out), and a fingerprint standing in for each module's build.
func capturedBody(t *testing.T, node string) ([]byte, map[string]string) {
	t.Helper()
	f, err := os.Open("testdata/declaration-" + node + ".json.gz")
	if err != nil {
		t.Fatal(err)
	}
	defer f.Close()
	z, err := gzip.NewReader(f)
	if err != nil {
		t.Fatal(err)
	}
	raw, err := io.ReadAll(z)
	if err != nil {
		t.Fatal(err)
	}
	var d struct {
		Builds map[string]string `json:"builds"`
	}
	if err := json.Unmarshal(raw, &d); err != nil {
		t.Fatal(err)
	}
	fps := map[string]string{}
	for m, b := range d.Builds {
		fps[m] = sdkdelivery.SourcePrefix2 + fmt.Sprintf("%064x", []byte(b))[:64]
	}
	return raw, fps
}

func resourceIDs(t *testing.T, raws []json.RawMessage) []string {
	t.Helper()
	var out []string
	for _, r := range raws {
		var head struct{ ID string }
		if err := json.Unmarshal(r, &head); err != nil {
			t.Fatal(err)
		}
		out = append(out, head.ID)
	}
	return out
}

// 1945/AC1: for a captured node's declaration, the union of its module compositions and its own is the declaration's
// resources, each once; each module's at its fingerprint; a module with no fingerprint of the new form (the private
// network's mesh-wireguard, which nobody builds) rides in the node's own.
func TestTheCompositionsOfANodeAreItsDeclaration(t *testing.T) {
	for _, node := range []string{"ace", "g14"} {
		body, fps := capturedBody(t, node)
		var d struct {
			Resources []json.RawMessage `json:"resources"`
		}
		_ = json.Unmarshal(body, &d)
		comps, moved, _, err := compositionsOf(node, body, fps, nil, nil, nil)
		if err != nil {
			t.Fatalf("%s: %v", node, err)
		}
		if strings.Join(moved, ",") != "mesh-wireguard" {
			t.Errorf("%s: moved into the node's own %v, want mesh-wireguard", node, moved)
		}
		var union []string
		for key, c := range comps {
			_, module, fp, isNode, err := sdkdelivery.ParseComposedKey(key)
			if err != nil || (!isNode && fp != fps[module]) {
				t.Errorf("%s: %s is at %s (%v)", node, key, fp, err)
			}
			if _, err := sdkdelivery.EncodeComposed(c); err != nil {
				t.Errorf("%s: %s: %v", node, key, err)
			}
			union = append(union, resourceIDs(t, c.Resources)...)
		}
		want := resourceIDs(t, d.Resources)
		sort.Strings(union)
		sort.Strings(want)
		if strings.Join(union, " ") != strings.Join(want, " ") {
			t.Errorf("%s: the compositions hold %d resources, the declaration %d, or they differ", node, len(union), len(want))
		}
		own, _ := sdkdelivery.ComposedNodeKey(node)
		if ids := strings.Join(resourceIDs(t, comps[own].Resources), " "); !strings.Contains(ids, "mesh-wireguard.") {
			t.Errorf("%s: the node's own composition does not hold mesh-wireguard's resources: %s", node, ids)
		}
	}
}

// 1945/AC1: a composition is written signed by the controller's key with its digest beside it; written again
// unchanged it writes nothing; a changed input writes a new revision of that key alone, by compare-and-set.
func TestACompositionIsWrittenOnceAndAgainOnlyWhenItChanges(t *testing.T) {
	body, fps := capturedBody(t, "g14")
	comps, _, _, err := compositionsOf("g14", body, fps, nil, nil, nil)
	if err != nil {
		t.Fatal(err)
	}
	at, pub, _, _ := aPlace(t, "g14")
	wrote, err := writeNodeCompositions(context.Background(), at, comps)
	if err != nil || len(wrote) != len(comps) {
		t.Fatalf("first write: %d of %d (%v)", len(wrote), len(comps), err)
	}
	for key := range comps {
		e, err := at.bucket.Get(context.Background(), key)
		if err != nil {
			t.Fatal(err)
		}
		c, err := sdkdelivery.DecodeComposed(e.Value())
		if err != nil {
			t.Fatal(err)
		}
		digest, err := sdkdelivery.VerifyComposed(c, []ed25519.PublicKey{pub})
		if err != nil {
			t.Fatalf("%s: %v", key, err)
		}
		d, err := at.digests.Get(context.Background(), key)
		if err != nil {
			t.Fatal(err)
		}
		v, err := sdkdelivery.DecodeComposedDigest(d.Value())
		if err != nil || v.Digest != digest {
			t.Errorf("%s: the digest beside it is %q, the composition's %s (%v)", key, v.Digest, digest, err)
		}
	}
	writes := at.bucket.(*fakeComposedKV).writes + at.digests.(*fakeComposedKV).writes
	if again, err := writeNodeCompositions(context.Background(), at, comps); err != nil || len(again) != 0 {
		t.Errorf("written again unchanged: %v (%v)", again, err)
	}
	if now := at.bucket.(*fakeComposedKV).writes + at.digests.(*fakeComposedKV).writes; now != writes {
		t.Errorf("an unchanged write wrote %d values", now-writes)
	}

	key, _ := sdkdelivery.ComposedKey("g14", "fail2ban", fps["fail2ban"])
	changed := comps[key]
	changed.Data = true
	comps[key] = changed
	wrote, err = writeNodeCompositions(context.Background(), at, comps)
	if err != nil || len(wrote) != 1 || wrote[0] != key {
		t.Errorf("a changed composition wrote %v (%v), want %s alone", wrote, err, key)
	}
}

// 1945/AC1: a module's compositions outside its five newest fingerprints are deleted with their digests, unless its
// want or its have names one.
func TestCompositionsOutsideTheFiveNewestAreDeleted(t *testing.T) {
	at, _, wants, haves := aPlace(t, "ace")
	file := json.RawMessage(`{"id":"m.f","type":"file","module":"m","path":"/etc/m","content":"x"}`)
	fp := func(i int) string { return sdkdelivery.SourcePrefix2 + strings.Repeat(fmt.Sprintf("%x", i), 64) }
	for i := 1; i <= 8; i++ {
		c := sdkdelivery.Composed{Node: "ace", Module: "m", FP: fp(i), Resources: []json.RawMessage{file}}
		key, _ := sdkdelivery.ComposedKey("ace", "m", fp(i))
		if _, err := writeNodeCompositions(context.Background(), at, map[string]sdkdelivery.Composed{key: c}); err != nil {
			t.Fatal(err)
		}
		if i == 1 {
			wants["m"] = fp(1)
		}
		if i == 2 {
			haves["m"] = sdkdelivery.Have{FP: fp(2), Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK}
		}
	}
	if err := pruneCompositions(context.Background(), at); err != nil {
		t.Fatal(err)
	}
	for i := 1; i <= 8; i++ {
		key, _ := sdkdelivery.ComposedKey("ace", "m", fp(i))
		_, err := at.bucket.Get(context.Background(), key)
		_, derr := at.digests.Get(context.Background(), key)
		kept := err == nil && derr == nil
		wantKept := i >= 4 || i == 1 || i == 2
		if kept != wantKept {
			t.Errorf("fingerprint %d kept %v, want %v (the wanted 1, the run 2, the five newest 4 to 8)", i, kept, wantKept)
		}
	}
}

// 1945 (found by the served test): a module the node runs that composes no resource of its own still has its
// composition, empty, at its fingerprint, so a want can name it; a module the node runs with no fingerprint of the new
// form gets none.
func TestAModuleWithNoResourceOfItsOwnStillHasItsComposition(t *testing.T) {
	body := []byte(`{"declaration":1,"resources":[]}`)
	fp := sdkdelivery.SourcePrefix2 + strings.Repeat("b", 64)
	comps, _, _, err := compositionsOf("laptop", body, map[string]string{"board": fp}, nil, []string{"board", "legacy"}, nil)
	if err != nil {
		t.Fatal(err)
	}
	key, _ := sdkdelivery.ComposedKey("laptop", "board", fp)
	c, ok := comps[key]
	if !ok || len(c.Resources) != 0 {
		t.Fatalf("the board's composition: %+v (%v)", c, ok)
	}
	if _, err := sdkdelivery.EncodeComposed(c); err != nil {
		t.Errorf("an empty composition does not encode: %v", err)
	}
	for k := range comps {
		if strings.Contains(k, "legacy") {
			t.Errorf("a module with no fingerprint of the new form was composed: %s", k)
		}
	}
}

// 1945/AC16 (unit): a module the controller knows no fingerprint of the new form for, whose want or have on the node
// names its own composition, is never in the node's own composition, neither its resources nor its filled paths, and is
// named as left; a module neither names still goes with the node's own, its filled path with it. (Live from 2026-10-25
// 12:43: zsh, power, restic and more were in both, and every node refused "2 resources declare X".)
func TestAModuleItsWantNamesIsNeverInTheNodesOwnComposition(t *testing.T) {
	body := []byte(`{"declaration":1,"resources":[
		{"id":"zsh.rc","type":"file","module":"zsh","path":"/etc/zsh/zshrc"},
		{"id":"power.unit","type":"file","module":"power","path":"/etc/power.conf"},
		{"id":"mesh-wireguard.conf","type":"file","module":"mesh-wireguard","path":"/etc/wireguard/mesh0.conf"}],
		"filled_files":["/etc/zsh/zshrc","/etc/power.conf","/etc/wireguard/mesh0.conf"]}`)
	asked := map[string]int{}
	named := func(module string) (bool, error) {
		asked[module]++
		return module == "zsh" || module == "power", nil
	}
	comps, moved, left, err := compositionsOf("ace", body, map[string]string{}, nil, nil, named)
	if err != nil {
		t.Fatal(err)
	}
	if strings.Join(left, ",") != "power,zsh" || strings.Join(moved, ",") != "mesh-wireguard" {
		t.Errorf("left %v, moved %v; want power,zsh left and mesh-wireguard moved", left, moved)
	}
	if asked["zsh"] != 1 || asked["power"] != 1 {
		t.Errorf("asked each module once: %v", asked)
	}
	own, _ := sdkdelivery.ComposedNodeKey("ace")
	for key, c := range comps {
		for _, id := range resourceIDs(t, c.Resources) {
			if strings.HasPrefix(id, "zsh.") || strings.HasPrefix(id, "power.") {
				t.Errorf("%s holds %s, which the module's own composition declares", key, id)
			}
		}
		for _, p := range c.Filled {
			if p == "/etc/zsh/zshrc" || p == "/etc/power.conf" {
				t.Errorf("%s names %s as filled, which the module's own composition names", key, p)
			}
		}
	}
	if ids := strings.Join(resourceIDs(t, comps[own].Resources), " "); ids != "mesh-wireguard.conf" {
		t.Errorf("the node's own composition holds %q, want mesh-wireguard.conf alone", ids)
	}
	if f := strings.Join(comps[own].Filled, " "); f != "/etc/wireguard/mesh0.conf" {
		t.Errorf("the node's own composition names %q as filled, want mesh-wireguard's file alone", f)
	}

	if _, _, _, err := compositionsOf("ace", body, map[string]string{}, nil, nil,
		func(string) (bool, error) { return false, fmt.Errorf("the wants bucket is gone") }); err == nil {
		t.Errorf("a want that cannot be read composed the node anyway")
	}
}