mesh-catalog/modules/mesh-vault/cmd/mesh-vault/mint_integration_test.go

//go:build integration

package main

// The vault whole against a real bus (novox/hq ADR 0561 §1 to §4, issue 774's acceptance 12): the built binary, played
// over its real stdin and stdout as the node's tool runtime plays it (tools/call with the caller in _meta; mesh/state.*
// and mesh/publish answered from a real nats-server's key-value buckets and subjects), its store in a directory of its
// own. A want written to the grants part's state is minted with no lease asked (issue 774: the vault is
// the one issuer): the hash is in bus-accounts, own answers the value sealed to the node, the value is the controller's
// shape and its hash's password, account-issued is said; an adopted user is never minted.

import (
	"bufio"
	"context"
	"crypto/rand"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"golang.org/x/crypto/bcrypt"
	"golang.org/x/crypto/nacl/box"

	"git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshvault"
	"git.novox.be/novox/mesh-sdk/go/buscompose"
	"git.novox.be/novox/mesh-sdk/go/seal"
	"git.novox.be/novox/mesh-sdk/go/testbus"
)

// wire is one line between the runtime and the bundle (the SDK's stdio message).
type wire struct {
	ID     json.RawMessage `json:"id,omitempty"`
	Method string          `json:"method,omitempty"`
	Params json.RawMessage `json:"params,omitempty"`
	Result json.RawMessage `json:"result,omitempty"`
	Error  *struct {
		Code    int    `json:"code"`
		Message string `json:"message"`
	} `json:"error,omitempty"`
}

// player is the node's tool runtime as far as the vault needs it, over a real bus.
type player struct {
	t   *testing.T
	ctx context.Context
	nc  *nats.Conn
	js  jetstream.JetStream

	mu      sync.Mutex
	in      io.Writer
	next    int
	pending map[string]chan wire
	said    []string
	// ask answers a mesh/ask (a call of another module's tool); nil answers none, as a call with no responder.
	ask func(key string, body json.RawMessage) (any, error)

	// The assigner's placement as its owner's tool and views family say it (ADR 0615): the values, and the revision of the
	// last change, kept from the bucket.
	placeMu       sync.Mutex
	placeValues   map[string]json.RawMessage
	placeRevision uint64
	// answeredOnce closes answered at the vault's first view of the placement.
	answeredOnce sync.Once
	answered     chan struct{}

	// The families the vault subscribed to, and the events said on one before it did: held, as the module's consumer on
	// the bus holds them, and handed in order once it subscribes. A runtime that refused the subscription and dropped the
	// events left the vault's picture at its first view, so a placement written after that view never reached it, and
	// whether it did turned on which came first (the rotation test's flake, met twice).
	subMu      sync.Mutex
	subscribed map[string]bool
	held       []heldEvent
}

// heldEvent is an event said on a family the vault has not subscribed to yet.
type heldEvent struct {
	key  string
	body json.RawMessage
}

// event says an event of key to the vault as its runtime does, holding it until the vault subscribed to its family.
func (p *player) event(key string, body json.RawMessage) {
	p.subMu.Lock()
	defer p.subMu.Unlock()
	if !p.subscribed[key] {
		p.held = append(p.held, heldEvent{key: key, body: body})
		return
	}
	p.deliver(key, body)
}

// deliver sends one event to the vault; under subMu, so events reach it in the order they were said.
func (p *player) deliver(key string, body json.RawMessage) {
	p.mu.Lock()
	p.next++
	id := fmt.Sprintf("o-%d", p.next)
	p.mu.Unlock()
	p.send(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": map[string]any{
		"key": key, "node": "novox", "body": body}}})
}

// subscribe marks the vault's subscription to pattern (a family's key, as the vault names it) and hands it the events
// held for it, in order.
func (p *player) subscribe(pattern string) {
	p.subMu.Lock()
	defer p.subMu.Unlock()
	if p.subscribed == nil {
		p.subscribed = map[string]bool{}
	}
	p.subscribed[pattern] = true
	rest := p.held[:0]
	for _, e := range p.held {
		if e.key == pattern {
			p.deliver(e.key, e.body)
			continue
		}
		rest = append(rest, e)
	}
	p.held = rest
}

var placeCreated = time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC)

// followPlacement keeps the bucket of the placement as the assigner's owner does: listed once, then each change said on
// its views family to the vault. Call it before the player serves.
func (p *player) followPlacement() {
	p.t.Helper()
	kv, err := p.js.KeyValue(p.ctx, "mesh-delivery-assigner_placement")
	if err != nil {
		p.t.Fatal(err)
	}
	w, err := kv.WatchAll(p.ctx)
	if err != nil {
		p.t.Fatal(err)
	}
	p.placeValues = map[string]json.RawMessage{}
	p.answered = make(chan struct{})
	listed := make(chan struct{})
	go func() {
		current := true
		for e := range w.Updates() {
			if e == nil {
				current = false
				close(listed)
				continue
			}
			p.placeMu.Lock()
			deleted := e.Operation() != jetstream.KeyValuePut
			if deleted {
				delete(p.placeValues, e.Key())
			} else {
				p.placeValues[e.Key()] = append(json.RawMessage(nil), e.Value()...)
			}
			if !current {
				p.placeRevision++
				ev := map[string]any{"v": 1, "owner": "mesh-delivery-assigner", "state": "placement", "key": e.Key(),
					"revision": p.placeRevision, "created": placeCreated}
				if deleted {
					ev["deleted"] = true
				} else {
					ev["value"] = json.RawMessage(e.Value())
				}
				body, _ := json.Marshal(ev)
				p.event(PlacementViewsEvent, body)
			}
			p.placeMu.Unlock()
		}
	}()
	select {
	case <-listed:
	case <-time.After(10 * time.Second):
		p.t.Fatal("the placement was not listed")
	}
}

// firstViewAnswered waits until the vault's first view of the placement is answered, so what the test writes after it
// reaches the vault only as the owner's events.
func (p *player) firstViewAnswered() {
	p.t.Helper()
	select {
	case <-p.answered:
	case <-time.After(30 * time.Second):
		p.t.Fatal("the vault never asked its first view of the placement")
	}
}

// bucketOf is the bucket a state names: the vault's own by its local name, another's as `<owner>.<name>`.
func bucketOf(state string) string {
	if owner, name, ok := strings.Cut(state, "."); ok {
		return owner + "_" + name
	}
	return "mesh-vault_" + state
}

func (p *player) send(m map[string]any) {
	raw, _ := json.Marshal(m)
	p.mu.Lock()
	defer p.mu.Unlock()
	_, _ = p.in.Write(append(raw, '\n'))
}

func (p *player) answer(id json.RawMessage, result any, err error) {
	if err != nil {
		p.send(map[string]any{"id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
		return
	}
	p.send(map[string]any{"id": id, "result": result})
}

func (p *player) kv(state string) (jetstream.KeyValue, error) {
	return p.js.KeyValue(p.ctx, bucketOf(state))
}

// serve answers what the bundle asks, and hands the answers to what the test asked it.
func (p *player) serve(out io.Reader) {
	scanner := bufio.NewScanner(out)
	scanner.Buffer(make([]byte, 1<<20), 1<<24)
	for scanner.Scan() {
		var m wire
		if json.Unmarshal(scanner.Bytes(), &m) != nil {
			continue
		}
		if m.Method == "" {
			var key string
			_ = json.Unmarshal(m.ID, &key)
			p.mu.Lock()
			ch := p.pending[key]
			delete(p.pending, key)
			p.mu.Unlock()
			if ch != nil {
				ch <- m
			}
			continue
		}
		go p.handle(m)
	}
}

func (p *player) handle(m wire) {
	var a struct {
		State string          `json:"state"`
		Key   string          `json:"key"`
		Value json.RawMessage `json:"value"`
		Body  json.RawMessage `json:"body"`
	}
	_ = json.Unmarshal(m.Params, &a)
	switch m.Method {
	case "mesh/state.get":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		e, err := kv.Get(p.ctx, a.Key)
		if errors.Is(err, jetstream.ErrKeyNotFound) {
			p.answer(m.ID, nil, nil)
			return
		}
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		p.answer(m.ID, map[string]any{"key": e.Key(), "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil)
	case "mesh/state.put":
		kv, err := p.kv(a.State)
		if err == nil {
			var rev uint64
			rev, err = kv.Put(p.ctx, a.Key, a.Value)
			if err == nil {
				p.answer(m.ID, map[string]any{"revision": rev}, nil)
				return
			}
		}
		p.answer(m.ID, nil, err)
	case "mesh/state.keys":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		keys, err := kv.Keys(p.ctx)
		if errors.Is(err, jetstream.ErrNoKeysFound) {
			keys, err = []string{}, nil
		}
		p.answer(m.ID, keys, err)
	case "mesh/state.delete":
		kv, err := p.kv(a.State)
		if err == nil {
			err = kv.Delete(p.ctx, a.Key)
		}
		p.answer(m.ID, map[string]any{}, err)
	case "mesh/state.watch":
		kv, err := p.kv(a.State)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		key := strings.ReplaceAll(a.Key, "**", ">")
		if key == "" {
			key = ">"
		}
		w, err := kv.Watch(p.ctx, key)
		if err != nil {
			p.answer(m.ID, nil, err)
			return
		}
		var watch string
		_ = json.Unmarshal(m.ID, &watch)
		current := true
		for e := range w.Updates() {
			if e == nil {
				current = false
				p.answer(m.ID, map[string]any{}, nil)
				continue
			}
			op := "put"
			if e.Operation() != jetstream.KeyValuePut {
				op = "delete"
			}
			change := map[string]any{"state": a.State, "key": e.Key(), "op": op, "revision": e.Revision(), "current": current}
			if op == "put" {
				change["value"] = json.RawMessage(e.Value())
			}
			p.mu.Lock()
			p.next++
			id := fmt.Sprintf("w-%d", p.next)
			p.mu.Unlock()
			p.send(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": change}})
		}
	case "mesh/publish":
		var e struct {
			Key  string          `json:"key"`
			Body json.RawMessage `json:"body"`
		}
		_ = json.Unmarshal(m.Params, &e)
		p.mu.Lock()
		p.said = append(p.said, e.Key+" "+string(e.Body))
		p.mu.Unlock()
		p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod.mesh-vault.event."+e.Key, e.Body))
	case "mesh/subscribe":
		var sub struct {
			Pattern string `json:"pattern"`
		}
		_ = json.Unmarshal(m.Params, &sub)
		p.answer(m.ID, map[string]any{}, nil)
		p.subscribe(sub.Pattern)
	case "mesh/ask":
		var ask struct {
			Key  string          `json:"key"`
			Body json.RawMessage `json:"body"`
		}
		_ = json.Unmarshal(m.Params, &ask)
		p.placeMu.Lock()
		following := p.placeValues != nil
		p.placeMu.Unlock()
		if ask.Key == PlacementTool && following {
			p.placeMu.Lock()
			answer := map[string]any{"v": 1, "owner": "mesh-delivery-assigner", "state": "placement", "revision": p.placeRevision,
				"created": placeCreated, "values": p.placeValues, "at": time.Now()}
			raw, _ := json.Marshal(answer)
			p.placeMu.Unlock()
			p.answer(m.ID, json.RawMessage(raw), nil)
			p.answeredOnce.Do(func() { close(p.answered) })
			return
		}
		if p.ask != nil {
			result, err := p.ask(ask.Key, ask.Body)
			p.answer(m.ID, result, err)
			return
		}
		p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
	default:
		// the rest: nothing answers on this bus, as a call with no responder.
		p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
	}
}

// call is a tools/call of the vault as caller, answered as the tool's JSON.
func (p *player) call(name, caller string, args map[string]any) (json.RawMessage, error) {
	p.mu.Lock()
	p.next++
	id := fmt.Sprintf("c-%d", p.next)
	ch := make(chan wire, 1)
	p.pending[id] = ch
	p.mu.Unlock()
	p.send(map[string]any{"id": id, "method": "tools/call", "params": map[string]any{"name": name, "arguments": args,
		"_meta": map[string]any{"mesh/caller": caller}}})
	select {
	case m := <-ch:
		var r struct {
			Content []struct {
				Text string `json:"text"`
			} `json:"content"`
			IsError bool `json:"isError"`
		}
		if m.Error != nil {
			return nil, errors.New(m.Error.Message)
		}
		if err := json.Unmarshal(m.Result, &r); err != nil || len(r.Content) != 1 {
			return nil, fmt.Errorf("%s answered %s", name, m.Result)
		}
		if r.IsError {
			return nil, errors.New(r.Content[0].Text)
		}
		return json.RawMessage(r.Content[0].Text), nil
	case <-time.After(30 * time.Second):
		return nil, fmt.Errorf("%s did not answer in 30s", name)
	}
}

func (p *player) put(t *testing.T, state, key string, v any) {
	t.Helper()
	kv, err := p.kv(state)
	if err != nil {
		t.Fatal(err)
	}
	raw, _ := json.Marshal(v)
	if _, err := kv.Put(p.ctx, key, raw); err != nil {
		t.Fatal(err)
	}
}

func (p *player) hashOf(t *testing.T, user string) *AccountHash {
	t.Helper()
	kv, err := p.kv(BusAccountsState)
	if err != nil {
		t.Fatal(err)
	}
	e, err := kv.Get(p.ctx, user)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil
	}
	if err != nil {
		t.Fatal(err)
	}
	var a AccountHash
	if err := json.Unmarshal(e.Value(), &a); err != nil {
		t.Fatal(err)
	}
	return &a
}

// strictly decodes raw into v, a field the type does not know being a fault (the definition's type is the contract).
func strictly(t *testing.T, raw []byte, v any) {
	t.Helper()
	dec := json.NewDecoder(strings.NewReader(string(raw)))
	dec.DisallowUnknownFields()
	if err := dec.Decode(v); err != nil {
		t.Errorf("%s does not match the seat's definition: %v", raw, err)
	}
}

func eventually(t *testing.T, what string, ok func() bool) {
	t.Helper()
	for end := time.Now().Add(30 * time.Second); time.Now().Before(end); time.Sleep(50 * time.Millisecond) {
		if ok() {
			return
		}
	}
	t.Fatalf("never: %s", what)
}

// 774/AC12.
func TestTheVaultWholeMintsOnTheBusWhoeverHoldsTheLease(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	dir := t.TempDir()
	bin := filepath.Join(dir, "mesh-vault")
	build := exec.Command("go", "build", "-o", bin, ".")
	build.Stderr = os.Stderr
	if err := build.Run(); err != nil {
		t.Fatalf("the vault does not build: %v", err)
	}
	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)
	}
	for _, b := range []string{"mesh-vault_bus-accounts", "mesh-grants_accounts",
		"mesh-provisioner_plans", "node-engine_have", "mesh-delivery-assigner_placement"} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 2}); err != nil {
			t.Fatal(err)
		}
	}
	// The operator key, as the mesh exports it to the vault's root.
	opPub, _, err := box.GenerateKey(rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	root := filepath.Join(dir, "root")
	_ = os.MkdirAll(root, 0o700)
	export, _ := json.Marshal(map[string]string{"operator-key": base64.StdEncoding.EncodeToString(opPub[:])})
	if err := os.WriteFile(filepath.Join(root, "export.json"), export, 0o600); err != nil {
		t.Fatal(err)
	}
	// An own secret of dunst on g14 is held before the vault starts, for the retire case below (CAT-V1).
	seedStore, err := OpenStore(filepath.Join(dir, "store"))
	if err != nil {
		t.Fatal(err)
	}
	seedKey, err := OwnKey(filepath.Join(dir, "state"), seedStore)
	if err != nil {
		t.Fatal(err)
	}
	seed := &Vault{Store: seedStore, Own: seedKey, Now: time.Now, Log: t.Logf, HashCost: bcrypt.MinCost}
	if _, err := seed.SetOperatorKey(asPEM(t, base64.StdEncoding.EncodeToString(opPub[:])), "the test"); err != nil {
		t.Fatal(err)
	}
	if _, err := seed.keep("own/g14/dunst/token", "a-token-value", "the test", "g14.dunst", OriginMade); err != nil {
		t.Fatal(err)
	}
	callers := filepath.Join(dir, "retire-callers.json")
	if err := os.WriteFile(callers, []byte(`{"callers":["mesh-config"]}`), 0o600); err != nil {
		t.Fatal(err)
	}
	cmd := exec.CommandContext(ctx, bin)
	cmd.Env = append(os.Environ(), "MESH_VAULT_STORE="+filepath.Join(dir, "store"), "MESH_VAULT_STATE="+filepath.Join(dir, "state"),
		"MESH_VAULT_ROOT="+root, "MESH_VAULT_LEDGER="+filepath.Join(dir, "ledger"), "MESH_RECEIVES="+filepath.Join(dir, "mesh.json"),
		"MESH_VAULT_RETIRE_CALLERS="+callers, "MESH_NODE=novox", "MESH_MODULE=mesh-vault")
	stdin, _ := cmd.StdinPipe()
	stdout, _ := cmd.StdoutPipe()
	cmd.Stderr = os.Stderr
	if err := cmd.Start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = stdin.Close(); cancel(); _ = cmd.Wait() })
	p := &player{t: t, ctx: ctx, nc: nc, js: js, in: stdin, pending: map[string]chan wire{}}
	// mesh-hand-acts holds the acts mesh-config recorded: one warranting the token's retirement, one for another name.
	p.ask = func(key string, _ json.RawMessage) (any, error) {
		if key != "mesh-hand-acts.hand-acts" {
			return nil, errors.New("nats: no responders available for request")
		}
		return map[string]any{"acts": []HandActSeen{
			{ID: "act-ok", By: "novox.mesh-config", At: time.Now(), Verb: "mesh-vault.retire", Args: []string{"g14 dunst token ret-1"}},
			{ID: "act-other", By: "novox.mesh-config", At: time.Now(), Verb: "mesh-vault.retire", Args: []string{"g14 dunst other ret-1"}}}}, nil
	}
	p.followPlacement()
	go p.serve(stdout)

	status := func() MintState {
		raw, err := p.call("vault_status", "person.jochen", nil)
		if err != nil {
			return MintState{}
		}
		var s struct {
			Minting MintState `json:"minting"`
		}
		_ = json.Unmarshal(raw, &s)
		return s.Minting
	}
	// No lease key: the want is minted all the same (issue 1126).
	x, y := aWant("mod-x"), aWant("mod-y")
	p.put(t, "mesh-vault.bus-accounts", y.User, AccountHash{User: y.User, Hash: "$2a$04$adoptedadoptedadoptedadoptedadoptedadoptedadopte", Adopted: true})
	p.put(t, "mesh-grants.accounts", x.User, x)
	p.put(t, "mesh-grants.accounts", y.User, y)
	eventually(t, "the account minted", func() bool { return p.hashOf(t, x.User) != nil })
	if st := status(); !st.On || len(st.HeldBack) != 0 {
		t.Errorf("vault_status says %+v", st)
	}
	nodePub, nodePriv := aSealingKey(t)
	if _, err := p.call(Seat+".node-key", "node.ace", map[string]any{"key": nodePub}); err != nil {
		t.Fatal(err)
	}
	raw, err := p.call(Seat+".own", "node.ace", map[string]any{})
	if err != nil {
		t.Fatal(err)
	}
	var own OwnAnswer
	if err := json.Unmarshal(raw, &own); err != nil {
		t.Fatal(err)
	}
	if len(own.Own) != 1 || own.Own[0].Module != "mod-x" || own.Own[0].Name != "broker" || own.Own[0].Origin != OriginMade {
		t.Fatalf("own answers %+v", own.Own)
	}
	value, err := seal.Open(own.Own[0].Sealed, nodePriv)
	if err != nil {
		t.Fatal(err)
	}
	var held struct{ Password string }
	_ = json.Unmarshal([]byte(value), &held)
	shape, _ := buscompose.HeldCredential(x, held.Password)
	if value != string(shape) {
		t.Errorf("the minted value %q is not the controller's shape %q", value, shape)
	}
	if h := p.hashOf(t, x.User); bcrypt.CompareHashAndPassword([]byte(h.Hash), []byte(held.Password)) != nil || h.Adopted || h.Generation != 1 {
		t.Errorf("bus-accounts %s holds %+v, not the minted password's hash", x.User, h)
	}
	if h := p.hashOf(t, y.User); h == nil || !h.Adopted {
		t.Errorf("the adopted %s was replaced: %+v", y.User, h)
	}
	p.mu.Lock()
	said := strings.Join(p.said, "\n")
	p.mu.Unlock()
	if !strings.Contains(said, AccountIssuedEvent+` {"`) || !strings.Contains(said, `"user":"ace.mod-x"`) || strings.Contains(said, held.Password) {
		t.Errorf("the vault said: %s", said)
	}

	// CAT-V1 (design 32): retire on the real bus. A waiter on rotated and own-replaced hears nothing of it.
	waiter, err := nc.SubscribeSync("mesh.mod.mesh-vault.event.>")
	if err != nil {
		t.Fatal(err)
	}
	for _, c := range []struct{ caller, act string }{{"g14.claude-code", "act-ok"}, {"novox.mesh-config", "act-other"}, {"novox.mesh-config", ""}} {
		if _, err := p.call(Seat+".retire", c.caller, map[string]any{"node": "g14", "module": "dunst", "name": "token", "warrant": c.act}); err == nil {
			t.Errorf("%s retired the token with act %q", c.caller, c.act)
		}
	}
	if m, err := waiter.NextMsg(300 * time.Millisecond); err == nil {
		t.Fatalf("a refused retire said %s", m.Subject)
	}
	// The call is the definition's input, the answer its answer, and the event its body on its subject (CAT-V4).
	input, _ := json.Marshal(meshvault.RetireInput{Node: "g14", Module: "dunst", Name: "token", Warrant: "act-ok", Ask: "ret-1"})
	var args map[string]any
	_ = json.Unmarshal(input, &args)
	raw, err = p.call(Seat+"."+meshvault.VerbRetire, "novox.mesh-config", args)
	if err != nil {
		t.Fatalf("retire with its warrant: %v", err)
	}
	var answered meshvault.RetireAnswer
	strictly(t, raw, &answered)
	if !answered.Retired {
		t.Errorf("retire answered %s", raw)
	}
	m, err := waiter.NextMsg(10 * time.Second)
	if err != nil || m.Subject != meshvault.HolderSubjectOwnRetired("mesh-vault") {
		t.Fatalf("retire said %v, %v", m, err)
	}
	var event meshvault.OwnRetiredHolderEvent
	strictly(t, m.Data, &event)
	if event.Ask != "ret-1" || event.Caller != "novox.mesh-config" || event.Node != "g14" || event.Module != "dunst" || event.Name != "token" {
		t.Errorf("own-retired said %s", m.Data)
	}
	for _, other := range []string{"rotated", OwnReplacedEvent} {
		if extra, err := waiter.NextMsg(300 * time.Millisecond); err == nil {
			t.Errorf("a waiter on %s heard %s", other, extra.Subject)
		}
	}
}