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 bus state over the
// in-memory mesh frombustest gives (restic on n1, requiring postgres-database from postgres there). 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"

	"git.novox.be/novox/mesh-sdk/go/asks"
	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"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/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)
	}
	a := &asked{}
	h := &Holder{Store: s, Asks: a, Now: time.Now, Logf: t.Logf,
		Compose: func(ctx context.Context, node string, keptAt map[string]string) (map[string]catalogue.Plan, map[string]string, error) {
			return frombus.PlansHeld(ctx, mesh, frombus.Mesh{}, frombus.Scope{Node: node, KeptAt: keptAt})
		}}
	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)
	}
	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
	if err := s.Delete(WantsRead, "want."+node+".restic"); err != nil {
		t.Fatal(err)
	}
	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)
	}
}