mesh-catalog/modules/disk-load/cmd/disk-load/whole_integration_test.go

//go:build integration

package main

// disk-load whole against a real bus (novox/hq ADR 0577 clause 5, the group's integration test): the built binary, played
// over its real stdin and stdout as the node's tool runtime plays it (mesh/state.* answered from a real nats-server's
// key-value buckets, a watch handed over from the bus's own watch, mesh/publish onto its subjects). A key of the
// node-engine's state data written after disk-load started has its module's item read within a minute, and disk-load
// composes nothing to learn it.

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/catalogue"
	"git.novox.be/novox/mesh-sdk/go/delivery"
	"git.novox.be/novox/mesh-sdk/go/health"
	"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"`
}

// player is the node's tool runtime, as far as disk-load 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
	asked []string // every state call, `<method> <state> <key>`
}

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{"jsonrpc": "2.0", "id": id, "error": map[string]any{"code": -32000, "message": err.Error()}})
		return
	}
	p.send(map[string]any{"jsonrpc": "2.0", "id": id, "result": result})
}

// bucket is a state's bucket: another module's `<owner>.<name>`, or disk-load's own by its local name.
func bucket(state string) string {
	if owner, name, other := strings.Cut(state, "."); other {
		return owner + "_" + name
	}
	return "disk-load_" + state
}

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 || m.Method == "" {
			continue // an answer to a change handed over
		}
		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)
	p.mu.Lock()
	p.asked = append(p.asked, m.Method+" "+a.State+" "+a.Key)
	p.mu.Unlock()
	if m.Method == "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.disk-load.event."+e.Key, e.Body))
		return
	}
	if !strings.HasPrefix(m.Method, "mesh/state.") {
		p.answer(m.ID, nil, errors.New("nats: no responders available for request"))
		return
	}
	kv, err := p.js.KeyValue(p.ctx, bucket(a.State))
	if err != nil {
		p.answer(m.ID, nil, err)
		return
	}
	switch m.Method {
	case "mesh/state.get":
		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":
		rev, err := kv.Put(p.ctx, a.Key, a.Value)
		p.answer(m.ID, map[string]any{"revision": rev}, err)
	case "mesh/state.keys":
		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.watch":
		p.watch(m.ID, a.State, kv, a.Key)
	default:
		p.answer(m.ID, nil, errors.New("not played here"))
	}
}

// watch hands over the bus's watch of the pattern: every current value, then the answer to the watch, then each change.
func (p *player) watch(id json.RawMessage, state string, kv jetstream.KeyValue, pattern string) {
	var watchID string
	_ = json.Unmarshal(id, &watchID)
	if pattern == "" {
		pattern = ">"
	}
	w, err := kv.Watch(p.ctx, strings.ReplaceAll(pattern, "**", ">"))
	if err != nil {
		p.answer(id, nil, err)
		return
	}
	current := true
	for e := range w.Updates() {
		if e == nil {
			current = false
			p.answer(id, map[string]any{}, nil)
			continue
		}
		op := "put"
		if e.Operation() != jetstream.KeyValuePut {
			op = "delete"
		}
		p.mu.Lock()
		p.next++
		n := p.next
		p.mu.Unlock()
		p.send(map[string]any{"jsonrpc": "2.0", "id": fmt.Sprintf("rt-%d", n), "method": "mesh/state", "params": map[string]any{
			"watch": watchID, "change": map[string]any{"state": state, "key": e.Key(), "op": op,
				"value": json.RawMessage(e.Value()), "revision": e.Revision(), "current": current}}})
	}
}

// 0577/AC5 (the group's integration test): a key of the node-engine's state data written after disk-load started is
// read within a minute, its item's reading is on the bus, and disk-load read no composition input to learn it.
func TestDiskLoadWholeMeasuresAnItemWithinAMinuteOfItsKey(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	t.Cleanup(cancel)
	bin := filepath.Join(t.TempDir(), "disk-load")
	build := exec.Command("go", "build", "-o", bin, ".")
	build.Stderr = os.Stderr
	if err := build.Run(); err != nil {
		t.Fatalf("disk-load 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{delivery.EngineDataBucket, health.ReadingsBucket, "disk-load_" + StateRounds} {
		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil {
			t.Fatal(err)
		}
	}
	cmd := exec.CommandContext(ctx, bin)
	cmd.Env = append(os.Environ(), "MESH_NODE=ace", "MESH_MODULE=disk-load")
	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}
	go p.serve(stdout)

	// Once disk-load watches its machine's keys, the engine writes one.
	watching := func() bool {
		p.mu.Lock()
		defer p.mu.Unlock()
		for _, a := range p.asked {
			if a == "mesh/state.watch "+delivery.EngineDataRead+" ace.*" {
				return true
			}
		}
		return false
	}
	for end := time.Now().Add(20 * time.Second); !watching(); time.Sleep(50 * time.Millisecond) {
		if time.Now().After(end) {
			t.Fatalf("disk-load did not watch its machine's keys of %s: asked %v", delivery.EngineDataBucket, p.asked)
		}
	}
	data, _ := js.KeyValue(ctx, delivery.EngineDataBucket)
	dir := t.TempDir()
	item := catalogue.PlacedDataItem{Module: "pg", ID: "data", Class: "valuable", Path: dir, Measure: catalogue.MeasureShallow}
	raw, err := delivery.EncodeEngineData(delivery.EngineData{Node: "ace", Module: "pg", Digest: "sha256:pg", Items: []catalogue.PlacedDataItem{item}})
	if err != nil {
		t.Fatal(err)
	}
	key, _ := delivery.EngineDataKey("ace", "pg")
	written := time.Now()
	if _, err := data.Put(ctx, key, raw); err != nil {
		t.Fatal(err)
	}
	readings, _ := js.KeyValue(ctx, health.ReadingsBucket)
	for {
		e, err := readings.Get(ctx, health.ReadingKey("ace", "pg", "data"))
		if err == nil {
			r, err := health.DecodeReading(e.Value())
			if err != nil || r.Path != dir || r.MeasuredAt == nil {
				t.Fatalf("the reading is %s (%v)", e.Value(), err)
			}
			break
		}
		if time.Since(written) > time.Minute {
			t.Fatalf("no reading within a minute of the key's write: %v", err)
		}
		time.Sleep(200 * time.Millisecond)
	}
	// It composed nothing: every state it asked is the engine's data or its own.
	p.mu.Lock()
	defer p.mu.Unlock()
	for _, a := range p.asked {
		f := strings.Fields(a)
		if len(f) > 1 && strings.HasPrefix(f[0], "mesh/state.") && f[1] != delivery.EngineDataRead && f[1] != StateReadings && f[1] != StateRounds {
			t.Errorf("disk-load read %s, which is no state of its own nor the engine's data", a)
		}
	}
}