mesh-catalog/modules/messenger/cmd/messenger/bus_test.go

package main

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

	"git.novox.be/novox/mesh-sdk/go/asks"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// The router started whole over a real bus (novox/hq ADR 0391, issue 926). A JetStream server is linked into the
// test and started for it alone; the module is this test binary started again as the bundle, speaking MCP on its
// standard streams to a runtime played here, which does what the tool runner does for it: its state is buckets
// of the bus, an ask submitted on the seat's subject is work handed to the bundle, what the bundle publishes on a
// seat is published on the bus, an event on the bus is handed to the bundle, and the controller's membership is
// the kinds this test says. It runs in the forge's merge check, not on the way.

const bundleVar = "MESH_MESSENGER_TEST_BUNDLE"

func TestMain(m *testing.M) {
	if os.Getenv(bundleVar) == "1" {
		kindsEvery = 200 * time.Millisecond
		main()
		os.Exit(0)
	}
	os.Exit(m.Run())
}

type busRuntime struct {
	t    *testing.T
	nc   *nats.Conn
	js   jetstream.JetStream
	in   io.WriteCloser
	mu   sync.Mutex // writes to the bundle, and the fields below
	next int
	wait map[string]chan json.RawMessage

	kinds     []map[string]any
	kindsRead int
	subscribed,
	taking bool
	buckets map[string]jetstream.KeyValue
}

func newBusRuntime(t *testing.T) *busRuntime {
	t.Helper()
	s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(30 * time.Second) {
		t.Fatal("no bus within 30s")
	}
	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
	nc, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	js, _ := jetstream.New(nc)
	return &busRuntime{t: t, nc: nc, js: js, wait: map[string]chan json.RawMessage{}, buckets: map[string]jetstream.KeyValue{}}
}

func (r *busRuntime) bucket(name string) jetstream.KeyValue {
	r.mu.Lock()
	defer r.mu.Unlock()
	if kv, ok := r.buckets[name]; ok {
		return kv
	}
	kv, err := r.js.CreateOrUpdateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: name})
	if err != nil {
		r.t.Fatal(err)
	}
	r.buckets[name] = kv
	return kv
}

func (r *busRuntime) setKinds(caps map[string][]string) {
	var out []map[string]any
	for kind, c := range caps {
		out = append(out, map[string]any{"seat": asks.ChannelSeat, "kind": kind, "module": kind, "node": "test", "capabilities": c})
	}
	r.mu.Lock()
	r.kinds = out
	r.mu.Unlock()
}

func (r *busRuntime) say(m map[string]any) {
	m["jsonrpc"] = "2.0"
	raw, _ := json.Marshal(m)
	r.mu.Lock()
	defer r.mu.Unlock()
	_, _ = r.in.Write(append(raw, '\n'))
}

// request is a request to the bundle (work, an event, a tool call) and its answer.
func (r *busRuntime) request(method string, params any) (json.RawMessage, error) {
	r.mu.Lock()
	r.next++
	id := "runtime-" + strconv.Itoa(r.next)
	ch := make(chan json.RawMessage, 1)
	r.wait[id] = ch
	r.mu.Unlock()
	r.say(map[string]any{"id": id, "method": method, "params": params})
	select {
	case raw := <-ch:
		return raw, nil
	case <-time.After(60 * time.Second):
		return nil, fmt.Errorf("%s was not answered in 60 s", method)
	}
}

func (r *busRuntime) start(dir string) {
	r.t.Helper()
	settings := filepath.Join(dir, "settings.json")
	_ = os.WriteFile(settings, []byte(`{}`), 0o600)
	cmd := exec.Command(os.Args[0], "-test.run=^$")
	cmd.Env = append(os.Environ(), bundleVar+"=1", "MESH_MESSENGER_SETTINGS="+settings, "MESH_MESSENGER_STATE="+dir,
		"MESH_SERVED_MODULE=messenger", "MESH_MODULE=messenger", "MESH_NODE=test")
	in, _ := cmd.StdinPipe()
	out, _ := cmd.StdoutPipe()
	cmd.Stderr = busLog{r.t}
	if err := cmd.Start(); err != nil {
		r.t.Fatal(err)
	}
	r.in = in
	go r.read(out)
	r.t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() })
}

type busLog struct{ t *testing.T }

func (l busLog) Write(p []byte) (int, error) { l.t.Logf("bundle: %s", p); return len(p), nil }

func (r *busRuntime) read(out io.Reader) {
	sc := bufio.NewScanner(out)
	sc.Buffer(make([]byte, 1<<20), 1<<24)
	for sc.Scan() {
		var m struct {
			ID     json.RawMessage `json:"id"`
			Method string          `json:"method"`
			Params json.RawMessage `json:"params"`
		}
		if json.Unmarshal(sc.Bytes(), &m) != nil {
			continue
		}
		if m.Method == "" {
			var id string
			_ = json.Unmarshal(m.ID, &id)
			r.mu.Lock()
			ch := r.wait[id]
			delete(r.wait, id)
			r.mu.Unlock()
			if ch != nil {
				ch <- append([]byte(nil), sc.Bytes()...)
			}
			continue
		}
		go r.runtime(m.ID, m.Method, m.Params)
	}
}

// runtime is what the tool runner does for the bundle: only what the router uses.
func (r *busRuntime) runtime(id json.RawMessage, method string, params json.RawMessage) {
	var p struct {
		State    string          `json:"state"`
		Key      string          `json:"key"`
		Value    json.RawMessage `json:"value"`
		Revision uint64          `json:"revision"`
		Body     json.RawMessage `json:"body"`
		Subject  string          `json:"subject"`
		Pattern  string          `json:"pattern"`
		Worker   string          `json:"worker"`
	}
	_ = json.Unmarshal(params, &p)
	ctx := context.Background()
	const changed = -32010
	code := -32000
	result, err := func() (any, error) {
		switch method {
		case "mesh/state.get":
			e, err := r.bucket(p.State).Get(ctx, p.Key)
			if errors.Is(err, jetstream.ErrKeyNotFound) {
				return nil, nil
			}
			if err != nil {
				return nil, err
			}
			return map[string]any{"key": p.Key, "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil
		case "mesh/state.put":
			rev, err := r.bucket(p.State).Put(ctx, p.Key, p.Value)
			return map[string]any{"revision": rev}, err
		case "mesh/state.create":
			rev, err := r.bucket(p.State).Create(ctx, p.Key, p.Value)
			if errors.Is(err, jetstream.ErrKeyExists) {
				code = changed
			}
			return map[string]any{"revision": rev}, err
		case "mesh/state.update":
			rev, err := r.bucket(p.State).Update(ctx, p.Key, p.Value, p.Revision)
			if err != nil {
				code = changed
			}
			return map[string]any{"revision": rev}, err
		case "mesh/state.delete":
			err := r.bucket(p.State).Delete(ctx, p.Key)
			return map[string]any{}, err
		case "mesh/state.keys":
			keys, err := r.bucket(p.State).Keys(ctx)
			if errors.Is(err, jetstream.ErrNoKeysFound) {
				return []string{}, nil
			}
			return keys, err
		case "mesh/subscribe":
			if strings.HasPrefix(p.Pattern, asks.IntakeSeat) {
				r.mu.Lock()
				r.subscribed = true
				r.mu.Unlock()
			}
			return map[string]any{}, nil
		case "mesh/seat.take":
			r.mu.Lock()
			r.taking = true
			r.mu.Unlock()
			return map[string]any{}, nil
		case "mesh/seat.answer", "mesh/publish":
			return map[string]any{}, nil
		case "mesh/seat.kinds":
			r.mu.Lock()
			defer r.mu.Unlock()
			r.kindsRead++
			return r.kinds, nil
		case "mesh/seat.publish":
			return map[string]any{}, r.nc.Publish(p.Subject, p.Body)
		case "mesh/ask":
			if p.Key == RootFreeVerb {
				var a struct {
					Machines []string `json:"machines"`
				}
				_ = json.Unmarshal(p.Body, &a)
				var out []RootVerdict
				for _, m := range a.Machines {
					out = append(out, RootVerdict{Machine: m, Free: true})
				}
				return map[string]any{"machines": out}, nil
			}
			return nil, fmt.Errorf("the test runtime answers nothing for %s", p.Key)
		}
		return nil, fmt.Errorf("the runtime offers no %s", method)
	}()
	if err != nil {
		r.say(map[string]any{"id": id, "error": map[string]any{"code": code, "message": err.Error()}})
		return
	}
	r.say(map[string]any{"id": id, "result": result})
}

// forward hands the bundle what is published on the bus that it takes: an ask submitted to the seat as work,
// a channel's word as an event.
func (r *busRuntime) forward() {
	if _, err := r.nc.Subscribe("mesh.seat."+asks.Seat+".accept.ask.*", func(m *nats.Msg) {
		_, _ = r.request("mesh/work", map[string]any{"work": map[string]any{"worker": AsksWorker, "subject": m.Subject, "body": json.RawMessage(m.Data)}})
	}); err != nil {
		r.t.Fatal(err)
	}
	if _, err := r.nc.Subscribe("mesh.seat."+asks.IntakeSeat+".event.>", func(m *nats.Msg) {
		key := strings.Replace(strings.TrimPrefix(m.Subject, "mesh.seat."), ".event", "", 1)
		_, _ = r.request("mesh/event", map[string]any{"envelope": map[string]any{"key": key, "node": "test", "body": json.RawMessage(m.Data)}})
	}); err != nil {
		r.t.Fatal(err)
	}
}

func (r *busRuntime) until(what string, within time.Duration, ok func() bool) {
	r.t.Helper()
	for deadline := time.Now().Add(within); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
		if ok() {
			return
		}
	}
	r.t.Fatalf("not within %s: %s", within, what)
}

func (r *busRuntime) askRecord(key string) *AskRecord {
	e, err := r.bucket("asks").Get(context.Background(), key)
	if err != nil {
		return nil
	}
	var rec AskRecord
	_ = json.Unmarshal(e.Value(), &rec)
	return &rec
}

// A held ask over the real bus (ADR 0397, issue 926): taken while the controller's word on Telegram lacks a
// verified sender, it is held with the reason and its asker is told nothing of an end; once the word gives it a
// verified sender, it is offered on Telegram.
func TestAnApprovalAskIsHeldThenOfferedWhenAChannelCanCarryItOverTheBus(t *testing.T) {
	r := newBusRuntime(t)
	away := []string{"deliver", "reaches-away", "choice", "edit", "exact-render"}
	r.setKinds(map[string][]string{"telegram": away})
	now := time.Now().UTC()
	// The operator's account, linked long enough ago to approve.
	id := Identity{Kind: "telegram", Identity: "42", Display: "Jo", Linked: now.Add(-3 * time.Hour), ApproveFrom: now.Add(-2 * time.Hour)}
	raw, _ := json.Marshal(id)
	if _, err := r.bucket("identities").Put(context.Background(), identityKey("telegram", "42"), raw); err != nil {
		t.Fatal(err)
	}
	var told, shown []json.RawMessage // shown is kept to say nothing was shown while held
	var seen sync.Mutex
	collect := func(into *[]json.RawMessage) nats.MsgHandler {
		return func(m *nats.Msg) { seen.Lock(); *into = append(*into, append([]byte(nil), m.Data...)); seen.Unlock() }
	}
	if _, err := r.nc.Subscribe(asks.DecidedSubject("mesh-delivery"), collect(&told)); err != nil {
		t.Fatal(err)
	}
	if _, err := r.nc.Subscribe(asks.ChannelSubject(asks.Show, "telegram"), collect(&shown)); err != nil {
		t.Fatal(err)
	}
	r.forward()
	r.start(t.TempDir())
	r.until("the bundle subscribed, takes asks and read the kinds", 20*time.Second, func() bool {
		r.mu.Lock()
		defer r.mu.Unlock()
		return r.subscribed && r.taking && r.kindsRead > 0
	})
	standing, _ := json.Marshal(Standing{Ready: true, At: time.Now().UTC()})
	if err := r.nc.Publish(asks.IntakeSubject(asks.StandingWhat, "telegram"), standing); err != nil {
		t.Fatal(err)
	}
	time.Sleep(time.Second)
	c := &clock{t: time.Now().UTC()}
	submitted, _ := json.Marshal(heldAsk(c, "d-1"))
	if err := r.nc.Publish(asks.AskSubject("mesh-delivery"), submitted); err != nil {
		t.Fatal(err)
	}
	r.until("the ask is recorded held with the reason", 20*time.Second, func() bool {
		rec := r.askRecord("mesh-delivery.d-1")
		return rec != nil && rec.State == StateHeld && strings.Contains(rec.HeldWhy, "Telegram cannot prove who answered")
	})
	rec := r.askRecord("mesh-delivery.d-1")
	if len(rec.Tickets) != 0 {
		t.Errorf("a held ask has tickets: %+v", rec)
	}
	seen.Lock()
	if len(told) != 0 || len(shown) != 0 {
		t.Errorf("a held ask was refused or shown: told %s, shown %s", told, shown)
	}
	seen.Unlock()
	// The asks verb shows it held, since when, and why.
	out, err := r.request("tools/call", map[string]any{"name": "operator-channel.asks", "arguments": map[string]any{}})
	if err != nil || !strings.Contains(string(out), `held_why`) || !strings.Contains(string(out), `held`) {
		t.Errorf("the asks verb: %s %v", out, err)
	}
	// The controller's word now gives Telegram a verified sender: the ask is offered there.
	r.setKinds(map[string][]string{"telegram": append(append([]string{}, away...), "verified-sender")})
	r.until("the ask is offered", 30*time.Second, func() bool {
		rec := r.askRecord("mesh-delivery.d-1")
		return rec != nil && rec.State == StateOpen && len(rec.Tickets) == 3 && rec.HeldWhy == ""
	})
	// Its message is the router's ordinary one for a new ask, said after the burst window (BurstWindow), which a
	// unit test covers with a clock; this test waits for no half minute.
	if rec := r.askRecord("mesh-delivery.d-1"); rec.Tickets == nil {
		t.Errorf("offered with no tickets: %+v", rec)
	}
}