mesh-catalog/modules/mesh-provisioner/cmd/mesh-provisioner/holder_integration_test.go

//go:build integration

package main

// The holder against a real bus (novox/hq ADR 0570 clauses 1-3, 5; issue 1954 H5): its states and every input it reads
// are a real nats-server's key-value buckets, its plans composed by mesh-sdk's composition from the watched copy of those
// buckets (ADR 0580 clause 2) filled with the mesh frombustest gives (restic on n1, requiring postgres-database from
// postgres there); no input is read key by key per composition. A placement and a
// want compose the key; the providers' outcomes (the vault's entry for the credential, postgres's own have) turn it
// provided; restic leaving the node makes it retiring and raises the ask.

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"testing"
	"time"

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

	stdio "git.novox.be/novox/mesh-sdk/go"
	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/compose/frombus"
	"git.novox.be/novox/mesh-sdk/go/compose/frombus/frombustest"
	"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/provisions"
	"git.novox.be/novox/mesh-sdk/go/testbus"
)

// kvStore is the Store over a real bus: this module's states by their local names, another's as `<module>.<name>`.
type kvStore struct {
	ctx context.Context
	js  jetstream.JetStream
}

func bucketOf(state string) string {
	if owner, name, ok := strings.Cut(state, "."); ok {
		return owner + "_" + name
	}
	return Module + "_" + state
}

func (s kvStore) kv(state string) (jetstream.KeyValue, error) {
	kv, err := s.js.KeyValue(s.ctx, bucketOf(state))
	if errors.Is(err, jetstream.ErrBucketNotFound) {
		return s.js.CreateKeyValue(s.ctx, jetstream.KeyValueConfig{Bucket: bucketOf(state)})
	}
	return kv, err
}

func (s kvStore) Get(state, key string) (*Entry, error) {
	kv, err := s.kv(state)
	if err != nil {
		return nil, err
	}
	e, err := kv.Get(s.ctx, key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &Entry{Key: key, Value: e.Value(), Revision: e.Revision()}, nil
}

func (s kvStore) Put(state, key string, value any) error {
	kv, err := s.kv(state)
	if err != nil {
		return err
	}
	raw, err := json.Marshal(value)
	if err != nil {
		return err
	}
	_, err = kv.Put(s.ctx, key, raw)
	return err
}

func (s kvStore) Delete(state, key string) error {
	kv, err := s.kv(state)
	if err != nil {
		return err
	}
	return kv.Delete(s.ctx, key)
}

func (s kvStore) Keys(state string) ([]string, error) {
	kv, err := s.kv(state)
	if err != nil {
		return nil, err
	}
	keys, err := kv.Keys(s.ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return nil, nil
	}
	return keys, err
}

type asked struct{ ids []string }

func (a *asked) Publish(_ string, body any, id string) error {
	a.ids = append(a.ids, id)
	_ = body.(asks.Ask)
	return nil
}
func (a *asked) Record(string) ([]byte, bool, error) { return nil, false, nil }

// 1954/H5.
func TestTheHolderComposesProvidesAndAsksOnTheBus(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	nc, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	s := kvStore{ctx: ctx, js: js}
	mesh, err := frombustest.Provisioned()
	if err != nil {
		t.Fatal(err)
	}
	node := frombustest.Node
	// The bus as the holder reads it: the indexer's modules and manifests, a placement naming where the vault runs, the
	// want of restic on n1.
	for module, m := range mesh.Module {
		m.V = 1
		raw, _ := json.Marshal(m)
		if err := s.Put(ModulesRead, module, json.RawMessage(raw)); err != nil {
			t.Fatal(err)
		}
	}
	for digest, raw := range mesh.Manifests {
		var named struct {
			Module string `json:"module"`
		}
		_ = json.Unmarshal(raw, &named)
		value, _ := delivery.EncodeManifest(delivery.Manifest{Module: named.Module, JSON: string(raw)})
		key, _ := delivery.ManifestsKey(digest)
		if err := s.Put(ManifestsRead, key, json.RawMessage(value)); err != nil {
			t.Fatal(err)
		}
	}
	// The vault keeps its outcomes as entries: its manifest declares `bound`.
	vaultManifest := `{"module":"mesh-vault","version":"1","state":["bound"]}`
	vaultDigest := delivery.ManifestDigest([]byte(vaultManifest))
	vaultValue, _ := delivery.EncodeManifest(delivery.Manifest{Module: provisions.VaultModule, JSON: vaultManifest})
	vaultKey, _ := delivery.ManifestsKey(vaultDigest)
	if err := s.Put(ManifestsRead, vaultKey, json.RawMessage(vaultValue)); err != nil {
		t.Fatal(err)
	}
	if err := s.Put(ModulesRead, provisions.VaultModule, map[string]any{"v": 1, "none2": "the test's", "manifest": vaultDigest,
		"repository": "novox/mesh-catalog", "commit": "c1"}); err != nil {
		t.Fatal(err)
	}
	vault, _ := json.Marshal(map[string]any{"v": 1, "node": "n2", "decided": map[string]any{},
		"modules": []map[string]any{{"module": provisions.VaultModule}}})
	if err := s.Put(PlacementRead, "placement.n2", json.RawMessage(vault)); err != nil {
		t.Fatal(err)
	}
	fp := frombustest.FPs()["restic"]
	if err := s.Put(WantsRead, "want."+node+".restic", map[string]any{"v": 1, "fp": fp, "self": true}); err != nil {
		t.Fatal(err)
	}
	// The rest of the mesh on the bus as its writers keep it: n1's placement, its record in its have, the builds.
	putPlacement := func() {
		raw, err := delivery.EncodePlacement(mesh.Placement[node])
		if err != nil {
			t.Fatal(err)
		}
		if err := s.Put(PlacementRead, "placement."+node, json.RawMessage(raw)); err != nil {
			t.Fatal(err)
		}
	}
	putPlacement()
	for _, n := range []string{node, "n2"} { // n2, where the vault runs, has its record too
		rec := mesh.Record[node]
		rec.Node = n
		recRaw, err := delivery.EncodeHaveNode(delivery.HaveNode{Record: &rec})
		if err != nil {
			t.Fatal(err)
		}
		if err := s.Put(HaveRead, mustKey(delivery.HaveNodeKey(n)), json.RawMessage(recRaw)); err != nil {
			t.Fatal(err)
		}
	}
	for at, b := range mesh.Builds {
		module, fp, _ := strings.Cut(at, " ")
		raw, err := delivery.EncodeBuilderBuild(*b)
		if err != nil {
			t.Fatal(err)
		}
		if err := s.Put(BuildsRead, mustKey(delivery.BuilderBuildKey(module, fp)), json.RawMessage(raw)); err != nil {
			t.Fatal(err)
		}
	}
	// The watched copy over the real buckets: each state by one watch, a build and a manifest by key.
	ws := stdio.WatchedStatesOver(func(state string, handle func(stdio.StateChange) error) error {
		kv, err := s.kv(state)
		if err != nil {
			return err
		}
		w, err := kv.WatchAll(ctx)
		if err != nil {
			return err
		}
		hand := func(e jetstream.KeyValueEntry) {
			op := stdio.OpPut
			if e.Operation() != jetstream.KeyValuePut {
				op = stdio.OpDelete
			}
			_ = handle(stdio.StateChange{State: state, Key: e.Key(), Op: op, Value: e.Value(), Revision: e.Revision()})
		}
		for e := range w.Updates() {
			if e == nil { // the current values are all handed over
				break
			}
			hand(e)
		}
		go func() {
			for e := range w.Updates() {
				hand(e)
			}
		}()
		return nil
	}, func(state, key string) ([]byte, bool, error) {
		e, err := s.Get(state, key)
		if err != nil || e == nil {
			return nil, false, err
		}
		return e.Value, true, nil
	})
	// settled waits until the copy holds a key as the bus does: in the mesh the holder is woken by that same watch.
	settled := func(state, key string) {
		t.Helper()
		for deadline := time.Now().Add(5 * time.Second); ; time.Sleep(10 * time.Millisecond) {
			e, _ := s.Get(state, key)
			v, found, _ := ws.Get(state, key)
			if (e == nil && !found) || (e != nil && found && string(e.Value) == string(v)) {
				return
			}
			if time.Now().After(deadline) {
				t.Fatalf("the copy of %s never took %s", state, key)
			}
		}
	}
	fetch := func(_ context.Context, url string) ([]byte, error) {
		for digest, body := range mesh.Archives {
			if strings.HasSuffix(url, digest) {
				return body, nil
			}
		}
		return nil, errors.New("no archive at " + url)
	}
	reader := frombus.StatesReader(fromruntime.Watched{WatchedStates: ws}, fetch)
	a := &asked{}
	h := &Holder{Store: Watching(s, ws), Asks: a, Now: time.Now, Logf: t.Logf,
		Compose: composerOver(reader)}
	state := func(key string) provisions.Provision {
		e, _ := s.Get(ProvisionsState, key)
		if e == nil {
			return provisions.Provision{}
		}
		p, _ := provisions.Decode(e.Value)
		return p
	}
	key := node + ".restic.postgres-database"
	if _, err := h.ComposeNode(ctx, node); err != nil {
		t.Fatal(err)
	}
	if p := state(key); p.State != provisions.Asked || p.Provider.Module != "postgres" {
		t.Fatalf("the placement and the want composed %+v", p)
	}
	if e, _ := s.Get(PlansState, node+".restic"); e == nil {
		t.Fatal("restic's plan is not on the bus")
	}
	// The providers' outcomes: the vault provides the credential where restic has one, postgres runs healthy.
	for _, dep := range state(key).DependsOn {
		if err := s.Put(boundRead(provisions.VaultModule), dep, map[string]any{"state": "provided"}); err != nil {
			t.Fatal(err)
		}
	}
	if err := s.Put(HaveRead, node+".postgres", map[string]any{"v": 1, "fp": frombustest.FPs()["postgres"], "phase": "running",
		"health": "ok"}); err != nil {
		t.Fatal(err)
	}
	settled(HaveRead, node+".postgres")
	for i := 0; i < 2; i++ { // the credential judged first, then what depends on it
		if _, err := h.ComposeNode(ctx, node); err != nil {
			t.Fatal(err)
		}
	}
	if p := state(key); p.State != provisions.Provided {
		t.Fatalf("the outcomes did not turn it provided: %+v", p)
	}
	// restic leaves the node and its want goes: retiring, one ask.
	p := mesh.Placement[node]
	var kept []delivery.PlacedModule
	for _, m := range p.Modules {
		if m.Module != "restic" {
			kept = append(kept, m)
		}
	}
	p.Modules = kept
	mesh.Placement[node] = p
	putPlacement()
	if err := s.Delete(WantsRead, "want."+node+".restic"); err != nil {
		t.Fatal(err)
	}
	settled(PlacementRead, "placement."+node)
	settled(WantsRead, "want."+node+".restic")
	if _, err := h.ComposeNode(ctx, node); err != nil {
		t.Fatal(err)
	}
	if r := state(key); r.State != provisions.Retiring || r.Ask == "" || len(a.ids) == 0 {
		t.Fatalf("removing the want did not ask: %+v, asked %v", r, a.ids)
	}
}