mesh-host/internal/compose/frombus/frombustest/frombustest_test.go

package frombustest

import (
	"context"
	"encoding/json"
	"reflect"
	"slices"
	"strings"
	"testing"

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/novox/mesh-host/internal/compose/frombus"
	"github.com/novox/mesh-host/internal/compose/fromruntime"
)

// ADR 0570 clause 2 (issue 1954; a3's H5): the in-memory mesh composes a plan from the bus that hands restic to the
// provisioner and names its postgres-database, provided by postgres on n1. And one bad module must not drop its node
// (issue 774): with a module placed beside it whose built manifest cannot be used, or whose plan refuses, PlansHeld
// still hands restic's plan and names the bad one with why, where Plans refuses the whole node. A manifest the indexer
// holds that the mesh refuses (power's bar item with attributes its bar places by none of, mesh-catalog ca8d6f33,
// 2026-10-30) holds that module alone on its node, in the plans and in the node's composition (Compose), and on a node
// it is not placed on holds nothing.
func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) {
	fp := "src2:" + strings.Repeat("9", 64)
	for _, row := range []struct {
		name string
		bad  string // the indexer's manifest of a module "backup" placed on n1; empty places none
		// built is the manifest backup's done build names; empty gives it no build
		built string
		why   string // words backup's hold says
		// elsewhere leaves backup in the indexer's modules and places it on no node: nothing is held.
		elsewhere bool
		// refused is the indexer's verdict on bad, a problem its judge found (novox/hq ADR 0578 §1): Plans then holds
		// backup too, and never ends the read.
		refused string
		// dataHeld says the data items' read holds backup too: a refusal it meets (a refused manifest, an unusable build,
		// an item composition refuses), not one only a plan meets.
		dataHeld bool
		// dataOnly checks the data items' read alone: the other reads blame restic, whose templates collect the item.
		dataOnly bool
	}{
		{name: "every module composes"},
		{name: "a built manifest that cannot be used",
			bad:   `{"module":"backup","version":"1","build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}`,
			built: `{"module":"backup","version":"1","requires":"amqp"}`, why: "backup", dataHeld: true},
		{name: "a plan that refuses",
			bad: `{"module":"backup","version":"1","resources":[{"id":"conf","type":"file","path":"/etc/backup.conf",` +
				`"mode":"0600","content":"key=${secret:nope}"}]}`, why: "${secret:nope}"},
		{name: "an indexer's manifest the mesh refuses", bad: unusableItem, why: `"shows" is none of them`,
			refused: `"shows" is none of them`, dataHeld: true},
		// ADR 0578 §3: an accepted manifest whose data item its composition refuses is held by the culprit rule, in every read.
		{name: "a data item its composition refuses",
			bad: `{"module":"backup","version":"1","data":{"own":[{"id":"d","path":"${dir:nope}","class":"cache"}]}}`,
			why: "nope", dataHeld: true, dataOnly: true},
		{name: "an indexer's manifest the mesh refuses, placed elsewhere", bad: unusableItem, elsewhere: true,
			refused: `"shows" is none of them`},
	} {
		b, err := Provisioned()
		if err != nil {
			t.Fatal(err)
		}
		if row.bad != "" {
			if !row.elsewhere {
				p := b.Placement[Node]
				p.Modules = append(p.Modules, delivery.PlacedModule{Module: "backup"})
				b.Placement[Node] = p
			}
			digest := delivery.ManifestDigest([]byte(row.bad))
			b.Manifests[digest] = []byte(row.bad)
			b.Module["backup"] = delivery.Module{FP2: fp, Manifest: digest}
			if row.refused != "" {
				b.Module["backup"] = delivery.Module{FP2: fp, Manifest: digest, Verdict: delivery.VerdictRefused,
					Problems: []string{row.refused}}
			}
			if row.built != "" {
				builtDigest := delivery.ManifestDigest([]byte(row.built))
				b.Manifests[builtDigest] = []byte(row.built)
				b.Builds["backup "+fp] = &delivery.BuilderBuild{State: delivery.BuildDone, Manifest: builtDigest,
					Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://backup/files/blobs/x", Digest: "sha256:x"}}}
			}
			if row.dataOnly {
				items, _, iheld, err := frombus.DataItemsHeld(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node})
				if why, ok := iheld["backup"]; err != nil || !ok || !strings.Contains(why, row.why) || len(items["restic"]) == 0 {
					t.Errorf("%s: data items %v held %v: %v", row.name, items, iheld, err)
				}
				continue
			}
			plans, err := frombus.Plans(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node})
			switch {
			case row.refused != "" && (err != nil || !planned(plans, "restic") || planned(plans, "backup")):
				t.Errorf("%s: Plans did not hold the refused manifest alone: %v %v", row.name, plans, err)
			case row.refused == "" && err == nil && !row.elsewhere:
				t.Errorf("%s: Plans composed the node; the row does not refuse it", row.name)
			}
		}
		if row.why != "" || row.bad == "" {
			// The node's own composition holds by the same rule: every other module composed, the bad one named.
			fps := FPs()
			if row.bad != "" {
				fps["backup"] = fp
			}
			out, _, held, err := frombus.Compose(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node}, fps)
			if err != nil {
				t.Fatalf("%s: the node's composition: %v", row.name, err)
			}
			if key, _ := delivery.ComposedKey(Node, "restic", fps["restic"]); out[key].Module != "restic" {
				t.Errorf("%s: restic is not composed (held %v)", row.name, held)
			}
			if why, ok := held["backup"]; (row.bad == "" && len(held) > 0) || (row.bad != "" && (!ok || !strings.Contains(why, row.why) || len(held) != 1)) {
				t.Errorf("%s: the node's composition held %v, want backup alone, saying %q", row.name, held, row.why)
			}
		}
		plans, held, err := frombus.PlansHeld(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node})
		if err != nil {
			t.Fatalf("%s: %v", row.name, err)
		}
		restic, handed := plans["restic"]
		if !handed || len(restic.Provisions) != 1 || restic.Provisions[0].Provision != "postgres-database" ||
			restic.Provisions[0].Provider != Node || restic.Provisions[0].ProviderModule != "postgres" {
			t.Fatalf("%s: the plans are %+v (held %v)", row.name, plans, held)
		}
		// Issue 1954: the plans at the wanted builds and at the newest, over one read, are what PlansHeld answers for each,
		// and where the time went is said by phase.
		keptAt := FPs()
		at, newest, took := frombus.PlansHeldTwice(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node, KeptAt: keptAt})
		wantAt, heldAt, errAt := frombus.PlansHeld(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node, KeptAt: keptAt})
		if !reflect.DeepEqual(at, frombus.HeldPlans{Plans: wantAt, Held: heldAt, Err: errAt}) ||
			!reflect.DeepEqual(newest, frombus.HeldPlans{Plans: plans, Held: held}) {
			t.Errorf("%s: twice composed %+v and %+v, want %v %v %v and %v %v", row.name, at, newest, wantAt, heldAt, errAt, plans, held)
		}
		if said := frombus.PhasesSaid(took); !strings.HasPrefix(said, "read ") || !strings.Contains(said, "plans at newest ") {
			t.Errorf("%s: the phases are said as %q", row.name, said)
		}
		if row.bad == "" || row.elsewhere {
			if len(held) > 0 {
				t.Errorf("%s: held %v", row.name, held)
			}
			continue
		}
		if why, ok := held["backup"]; !ok || !strings.Contains(why, row.why) || len(held) != 1 {
			t.Errorf("%s: held %v, want backup alone, saying %q", row.name, held, row.why)
		}
		if _, planned := plans["backup"]; planned {
			t.Errorf("%s: the held module has a plan", row.name)
		}
		// The data items are read by the same rule (ADR 0578 §3: disk-load's read): the bad module held, the read whole.
		items, _, iheld, err := frombus.DataItemsHeld(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node})
		// A refusal only a plan meets (a file's secret) holds nothing here: data items render no such file.
		mustHold := row.dataHeld
		if why, ok := iheld["backup"]; err != nil || ok != mustHold || mustHold && !strings.Contains(why, row.why) || items["backup"] != nil {
			t.Errorf("%s: data items %v held %v: %v", row.name, items, iheld, err)
		}
	}

	// ADR 0580 clause 1: the watched reader reads a state it composes from out of the copy, a build by key each time it is
	// asked, and a manifest once per digest.
	byKey := map[string]int{}
	ws := stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error {
		byKey[state+" watched"]++
		return handle(stdio.StateChange{Key: "placement." + Node, Op: stdio.OpPut, Value: []byte(`{}`), Current: true})
	}, func(state, key string) ([]byte, bool, error) { byKey[state]++; return []byte(key), true, nil })
	w := fromruntime.Watched{WatchedStates: ws}
	for range 2 {
		_, _, _ = w.Get(frombus.BuildsRead, "restic."+fp)
		_, _, _ = w.Get(frombus.ManifestsRead, "sha256-d")
		if _, found, _ := w.Get(frombus.PlacementRead, "placement."+Node); !found {
			t.Error("the watched reader does not answer the placement from its copy")
		}
	}
	if byKey[frombus.BuildsRead] != 2 || byKey[frombus.ManifestsRead] != 1 || byKey[frombus.PlacementRead] != 0 ||
		byKey[frombus.PlacementRead+" watched"] != 1 {
		t.Errorf("the watched reader read %v, want a build by key each time, a manifest once, the placement watched once", byKey)
	}
	fetched := 0
	keep := fromruntime.KeptFetch(func(context.Context, string) ([]byte, error) { fetched++; return []byte("tar"), nil })
	for range 2 {
		if body, err := keep(context.Background(), "http://store/restic/files/blobs/sha256:a"); err != nil || string(body) != "tar" {
			t.Fatalf("the kept fetch answered %q, %v", body, err)
		}
	}
	if fetched != 1 {
		t.Errorf("a templates archive was fetched %d times, want once", fetched)
	}

	// Issue 1954: the states reader takes a build in once per its content: composed again, its archive is not read again
	// and the plans are the same.
	b, err := Provisioned()
	if err != nil {
		t.Fatal(err)
	}
	reader, archives := frombus.StatesReader(statesOf(t, b), func(_ context.Context, url string) ([]byte, error) {
		fetched++
		_, digest, _ := strings.Cut(url, "/blobs/")
		return b.Archives[digest], nil
	}), fetched
	var first map[string]catalogue.Plan
	for i := range 2 {
		plans, held, err := frombus.PlansHeld(context.Background(), reader, frombus.Mesh{}, frombus.Scope{Node: Node})
		if err != nil || len(held) > 0 || (i == 1 && !reflect.DeepEqual(plans, first)) || !planned(plans, "restic") {
			t.Fatalf("composed %v (held %v): %v", plans, held, err)
		}
		first = plans
	}
	if fetched-archives != 1 {
		t.Errorf("restic's archive was read %d times over two compositions, want once", fetched-archives)
	}

	// Issue 774 (ace, 2026-11-04 01:29): the provisioner client held as its stub (its build not done, so the node keeps
	// running what it runs) keeps the account the node's own composition names, and restic stays handed to it with its
	// plan; without it every filled write was refused and no module was handed.
	if b, err = Provisioned(); err != nil {
		t.Fatal(err)
	}
	fps, ownKey := FPs(), ""
	ownKey, _ = delivery.ComposedNodeKey(Node)
	resticKey, _ := delivery.ComposedKey(Node, "restic", fps["restic"])
	before, _, _, err := frombus.Compose(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node}, fps)
	account := before[ownKey].Own.Provisioner
	if err != nil || account == "" || len(before[resticKey].Filled) == 0 {
		t.Fatalf("the node composes with its client: account %q, restic's filled files %v: %v", account, before[resticKey].Filled, err)
	}
	var running []delivery.Composed
	for _, c := range before {
		running = append(running, c)
	}
	for _, row := range []struct {
		own, placed bool // the account read from the running own composition, else the client's running process; the client placed
		userless    bool // the client placed and not held, its process naming no user: it composes no account
		// filledHeld holds every module with a filled file (restic, its build not done) and not the client: the node still
		// names the account (issue 774, ace 2026-11-04: its build modules and ssh-client held, no filled file, no account)
		filledHeld bool
	}{{true, true, false, false}, {false, true, false, false}, {true, false, false, false}, {false, false, false, false},
		{true, true, true, false}, {true, true, false, true}} {
		own := row.own
		runs := running
		if !own { // what runs names no account of its own: the client's running process says it
			runs = nil
			for _, c := range running {
				if c.Module == "" {
					c.Own.Provisioner = ""
				}
				runs = append(runs, c)
			}
		}
		bb, err := Provisioned()
		if err != nil {
			t.Fatal(err)
		}
		want := account
		if row.userless {
			// placed and not held, composing no account of its own: no stale account is kept from what runs
			mod := bb.Module["mesh-provisioner-client"]
			var m map[string]any
			if err := json.Unmarshal(bb.Manifests[mod.Manifest], &m); err != nil {
				t.Fatal(err)
			}
			for _, r := range m["resources"].([]any) {
				delete(r.(map[string]any), "user")
			}
			raw, _ := json.Marshal(m)
			mod.Manifest = delivery.ManifestDigest(raw)
			bb.Module["mesh-provisioner-client"], bb.Manifests[mod.Manifest] = mod, raw
			built := *bb.Builds["mesh-provisioner-client "+fps["mesh-provisioner-client"]]
			built.Manifest = mod.Manifest
			bb.Builds["mesh-provisioner-client "+fps["mesh-provisioner-client"]] = &built
			want = ""
		} else if row.filledHeld {
			bb.Builds["restic "+fps["restic"]] = &delivery.BuilderBuild{State: delivery.BuildBuilding}
		} else if row.placed {
			// held: its build not done while it runs
			bb.Builds["mesh-provisioner-client "+fps["mesh-provisioner-client"]] = &delivery.BuilderBuild{State: delivery.BuildBuilding}
		} else {
			// taken out of the node's placement while what runs still names it: the account goes with it, never kept stale
			p := bb.Placement[Node]
			p.Modules = slices.DeleteFunc(slices.Clone(p.Modules), func(m delivery.PlacedModule) bool { return m.Module == "mesh-provisioner-client" })
			bb.Placement[Node] = p
			want = ""
		}
		after, _, held, err := frombus.Compose(context.Background(), bb, frombus.Mesh{}, frombus.Scope{Node: Node, Running: runs}, fps)
		if _, isHeld := held["mesh-provisioner-client"]; err != nil || isHeld != (row.placed && !row.userless && !row.filledHeld) {
			t.Fatalf("%+v: held %v: %v", row, held, err)
		}
		if got := after[ownKey].Own.Provisioner; got != want {
			t.Errorf("%+v: the node's own names %q, want %q", row, got, want)
		}
		if _, resticHeld := held["restic"]; row.filledHeld && !resticHeld {
			t.Errorf("%+v: restic held %v", row, resticHeld)
		}
		if row.placed && !row.userless && !row.filledHeld && !reflect.DeepEqual(after[resticKey].Filled, before[resticKey].Filled) {
			t.Errorf("%+v: restic's filled files %v, want %v", row, after[resticKey].Filled, before[resticKey].Filled)
		}
	}
}

// statesOf is the bus b as a module's read states hold it.
func statesOf(t *testing.T, b *Bus) frombus.States {
	t.Helper()
	s := states{}
	put := func(state, key string, raw []byte, err error) {
		if err != nil {
			t.Fatal(err)
		}
		if s[state] == nil {
			s[state] = map[string][]byte{}
		}
		s[state][key] = raw
	}
	for node, p := range b.Placement {
		raw, err := delivery.EncodePlacement(p)
		put(frombus.PlacementRead, "placement."+node, raw, err)
	}
	for node, rec := range b.Record {
		raw, err := delivery.EncodeHaveNode(delivery.HaveNode{Record: &rec})
		put(frombus.HaveRead, node+".node", raw, err)
	}
	for module, m := range b.Module {
		key, _ := delivery.ModulesKey(module)
		raw, err := delivery.EncodeModule(m)
		put(frombus.ModulesRead, key, raw, err)
	}
	for digest, man := range b.Manifests {
		key, _ := delivery.ManifestsKey(digest)
		raw, err := delivery.EncodeManifest(delivery.Manifest{Module: "m", JSON: string(man)})
		put(frombus.ManifestsRead, key, raw, err)
	}
	for at, build := range b.Builds {
		module, fp, _ := strings.Cut(at, " ")
		key, _ := delivery.BuilderBuildKey(module, fp)
		raw, err := delivery.EncodeBuilderBuild(*build)
		put(frombus.BuildsRead, key, raw, err)
	}
	return s
}

// states are read states in memory.
type states map[string]map[string][]byte

func (s states) Get(state, key string) ([]byte, bool, error) {
	v, ok := s[state][key]
	return v, ok, nil
}

func (s states) Keys(state string) ([]string, error) {
	var keys []string
	for k := range s[state] {
		keys = append(keys, k)
	}
	return keys, nil
}

func planned[P any](plans map[string]P, module string) bool { _, ok := plans[module]; return ok }

// unusableItem is a manifest the mesh refuses as the node-engine did power's at mesh-catalog ca8d6f33: a bar block item
// with attributes its bar places by none of.
const unusableItem = `{"module":"backup","version":"1","moved":true,` +
	`"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]},` +
	`"items":[{"collection":"contribution:node-bar:block",` +
	`"attributes":{"bar":"bottom","place":"status","shows":"state","state":"power.draw"},"template":"files/bar"}]}`