mesh-catalog/modules/mesh-self-check/cmd/mesh-self-check/whole_integration_test.go

//go:build integration

package main

// The self-check whole against a real bus (novox/hq ADR 0564 §6, issue 1952 SC1): the built binary, played over its
// real stdin and stdout as the node's tool runtime plays it (tools/call; mesh/state.* answered from a real
// nats-server's key-value buckets, mesh/publish onto its subjects, and the bus module's nats_streams answered from
// that server's own streams). One run asked through the seat's verb finds the mesh's streams missing on a bare bus,
// writes them as findings into its bucket on the bus, keeps the verdict, and says checked.

import (
	"bufio"
	"context"
	"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"

	"git.novox.be/novox/mesh-sdk/go/testbus"
)

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 {
		Message string `json:"message"`
	} `json:"error,omitempty"`
}

// player is the node's tool runtime, as far as the self-check needs it, over a real bus.
type player struct {
	ctx     context.Context
	nc      *nats.Conn
	js      jetstream.JetStream
	mu      sync.Mutex
	in      io.Writer
	next    int
	pending map[string]chan wire
}

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) 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) kv(state string) (jetstream.KeyValue, error) {
	return p.js.KeyValue(p.ctx, Module+"_"+state)
}

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
			if rev, err = kv.Put(p.ctx, a.Key, a.Value); 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/publish":
		var e struct {
			Key  string          `json:"key"`
			Body json.RawMessage `json:"body"`
		}
		_ = json.Unmarshal(m.Params, &e)
		p.answer(m.ID, map[string]any{}, p.nc.Publish("mesh.mod."+Module+".event."+e.Key, e.Body))
	case "mesh/ask":
		var ask struct {
			Key string `json:"key"`
		}
		_ = json.Unmarshal(m.Params, &ask)
		if ask.Key != StreamsTool {
			p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
			return
		}
		// The bus module's answer, from the server's own streams.
		var detail []map[string]any
		for name := range p.js.StreamNames(p.ctx).Name() {
			detail = append(detail, map[string]any{"name": name})
		}
		p.answer(m.ID, map[string]any{"detail": detail}, nil)
	default:
		p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
	}
}

func (p *player) call(name 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": "person.jochen"}}})
	select {
	case m := <-ch:
		if m.Error != nil {
			return nil, errors.New(m.Error.Message)
		}
		var r struct {
			Content []struct {
				Text string `json:"text"`
			} `json:"content"`
			IsError bool `json:"isError"`
		}
		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(60 * time.Second):
		return nil, fmt.Errorf("%s did not answer in 60s", name)
	}
}

// 1952/SC1: one run of the self-check whole ends with checked and a verdict, its findings on the bus.
func TestTheSelfCheckWholeRunsOnTheBus(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	dir := t.TempDir()
	bin := filepath.Join(dir, Module)
	build := exec.Command("go", "build", "-o", bin, ".")
	build.Stderr = os.Stderr
	if err := build.Run(); err != nil {
		t.Fatalf("the self-check 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 _, s := range []string{FindingsState, VerdictState, VerdictsState} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: Module + "_" + s}); err != nil {
			t.Fatal(err)
		}
	}
	checked, err := nc.SubscribeSync("mesh.mod." + Module + ".event." + CheckedEvent)
	if err != nil {
		t.Fatal(err)
	}
	cmd := exec.CommandContext(ctx, bin)
	cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE="+Module)
	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{ctx: ctx, nc: nc, js: js, in: stdin, pending: map[string]chan wire{}}
	go p.serve(stdout)

	raw, err := p.call(DoctorTool, map[string]any{"run": true})
	if err != nil {
		t.Fatal(err)
	}
	var answer struct {
		Run Run `json:"run"`
	}
	if err := json.Unmarshal(raw, &answer); err != nil {
		t.Fatal(err)
	}
	verdicts := map[string]string{}
	for _, v := range answer.Run.Probes {
		verdicts[v.ID] = v.Verdict
	}
	if verdicts["D7"] != VerdictFail || verdicts["DW"] != VerdictFail || verdicts["D15"] != VerdictFailedToRun ||
		verdicts["D2"] != VerdictDeferred || answer.Run.Controller != "ace" || !strings.Contains(answer.Run.Why, "person.jochen") {
		t.Errorf("the run answered %+v", answer.Run)
	}
	msg, err := checked.NextMsg(10 * time.Second)
	if err != nil {
		t.Fatalf("checked was not said: %v", err)
	}
	var said Run
	if err := json.Unmarshal(msg.Data, &said); err != nil || said.Run != answer.Run.Run {
		t.Errorf("checked said %s", msg.Data)
	}
	findings, _ := js.KeyValue(ctx, Module+"_"+FindingsState)
	e, err := findings.Get(ctx, "bus.EVENTS.stream-missing")
	if err != nil {
		t.Fatalf("the missing stream is not a finding on the bus: %v", err)
	}
	var f Finding
	if err := json.Unmarshal(e.Value(), &f); err != nil || f.Kind != "stream-wrong" || f.Severity != Urgent || f.Source != "D7" {
		t.Errorf("the finding is %s", e.Value())
	}
	verdict, _ := js.KeyValue(ctx, Module+"_"+VerdictState)
	if _, err := verdict.Get(ctx, LastKey); err != nil {
		t.Errorf("the last verdict is not kept: %v", err)
	}
	if raw, err := p.call(VerdictsTool, map[string]any{}); err != nil || !strings.Contains(string(raw), answer.Run.Run) {
		t.Errorf("the week's verdicts answered %s, %v", raw, err)
	}
}