mesh-sdk/go/log_test.go

package stdio

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strings"
	"sync"
	"testing"
	"time"
)

// logRuntime is the runtime's side of mesh/log.append, read, last and info as the bus does them (novox/hq
// ADR 0297): one sequence per entry across the log, an append with expect refused unless the key's last
// entry is still at it (0 for a key with none), that refusal answered with CodeChanged as a lost state
// update is, and a read paged by its limit with the sequence to read on from. asked keeps every ask's
// method and params, as the bundle sent them.
type logRuntime struct {
	mu      sync.Mutex
	entries []LogEntry
	asked   []message
}

func (r *logRuntime) last(key string) uint64 {
	for i := len(r.entries) - 1; i >= 0; i-- {
		if r.entries[i].Key == key {
			return r.entries[i].Sequence
		}
	}
	return 0
}

// matches is a key pattern as the runtime reads it: `*` one name, `>` the rest.
func matches(pattern, key string) bool {
	if pattern == "" {
		return true
	}
	p, k := strings.Split(pattern, "."), strings.Split(key, ".")
	for i, t := range p {
		if t == ">" {
			return len(k) > i
		}
		if i >= len(k) || (t != "*" && t != k[i]) {
			return false
		}
	}
	return len(p) == len(k)
}

func (r *logRuntime) answer(m message) map[string]any {
	var p struct {
		Log    string          `json:"log"`
		Key    string          `json:"key"`
		Value  json.RawMessage `json:"value"`
		Expect *uint64         `json:"expect"`
		From   uint64          `json:"from"`
		Keys   string          `json:"keys"`
		Limit  int             `json:"limit"`
	}
	_ = json.Unmarshal(m.Params, &p)
	r.mu.Lock()
	defer r.mu.Unlock()
	r.asked = append(r.asked, m)
	if p.Log != "changes" {
		return map[string]any{"error": map[string]any{"code": -32000, "message": "issues was issued no log " + p.Log}}
	}
	switch strings.TrimPrefix(m.Method, "mesh/log.") {
	case "append":
		if p.Expect != nil && r.last(p.Key) != *p.Expect {
			return map[string]any{"error": map[string]any{"code": CodeChanged, "message": p.Key + ": the key was written by another writer first"}}
		}
		seq := uint64(len(r.entries)) + 1
		r.entries = append(r.entries, LogEntry{Sequence: seq, Key: p.Key, Value: p.Value,
			At: time.Date(2026, 10, 11, 18, 0, int(seq), 0, time.UTC)})
		return map[string]any{"result": map[string]any{"sequence": seq}}
	case "read":
		limit := p.Limit
		if limit == 0 {
			limit = 100
		}
		page := map[string]any{"entries": []LogEntry{}, "next": 0, "last": len(r.entries)}
		var got []LogEntry
		for _, e := range r.entries {
			if e.Sequence < p.From || !matches(p.Keys, e.Key) {
				continue
			}
			if len(got) == limit {
				page["next"] = e.Sequence
				break
			}
			got = append(got, e)
		}
		if got != nil {
			page["entries"] = got
		}
		return map[string]any{"result": page}
	case "last":
		seq := r.last(p.Key)
		if seq == 0 {
			return map[string]any{"result": nil}
		}
		return map[string]any{"result": r.entries[seq-1]}
	case "info":
		var bytes int
		for _, e := range r.entries {
			bytes += len(e.Value)
		}
		info := map[string]any{"entries": len(r.entries), "bytes": bytes, "max_bytes": 1 << 30, "first": 0, "last": len(r.entries)}
		if len(r.entries) > 0 {
			info["first"] = 1
		}
		return map[string]any{"result": info}
	}
	return map[string]any{"error": map[string]any{"code": -32601, "message": "no " + m.Method}}
}

// logging serves a bundle whose one tool, "log", runs run against r, answering each mesh/log ask beside
// the loop as the runtime does. It returns what the tool answered.
func logging(t *testing.T, r *logRuntime, run func() (any, error)) map[string]any {
	t.Helper()
	t.Setenv(ServedAs, "issues")
	inR, inW := io.Pipe()
	outR, outW := io.Pipe()
	go func() {
		_ = ServeOn("", []Tool{{Name: "log", Run: func(map[string]any) (any, error) { return run() }}}, inR, outW)
	}()
	var wmu sync.Mutex
	send := func(m map[string]any) {
		m["jsonrpc"] = "2.0"
		b, _ := json.Marshal(m)
		wmu.Lock()
		defer wmu.Unlock()
		_, _ = inW.Write(append(b, '\n'))
	}
	done := make(chan map[string]any, 1)
	go func() {
		lines := bufio.NewScanner(outR)
		lines.Buffer(make([]byte, 1<<20), 1<<24)
		for lines.Scan() {
			var m message
			if json.Unmarshal(lines.Bytes(), &m) != nil {
				continue
			}
			if strings.HasPrefix(m.Method, "mesh/log.") {
				go func(m message) {
					a := r.answer(m)
					a["id"] = m.ID
					send(a)
				}(m)
				continue
			}
			var answer map[string]any
			_ = json.Unmarshal(lines.Bytes(), &answer)
			if answer["id"] == float64(1) {
				done <- answer
				return
			}
		}
	}()
	send(map[string]any{"id": 1, "method": "tools/call", "params": map[string]any{"name": "log"}})
	t.Cleanup(func() { _ = inW.Close() })
	select {
	case a := <-done:
		return a
	case <-time.After(10 * time.Second):
		t.Fatal("the bundle did not answer")
		return nil
	}
}

// params is what the bundle sent with its n-th ask, and the ask's method.
func (r *logRuntime) params(t *testing.T, n int) (string, map[string]any) {
	t.Helper()
	r.mu.Lock()
	defer r.mu.Unlock()
	if n >= len(r.asked) {
		t.Fatalf("the bundle asked %d times, not %d", len(r.asked), n+1)
	}
	var p map[string]any
	_ = json.Unmarshal(r.asked[n].Params, &p)
	return r.asked[n].Method, p
}

// A bundle appends to, reads, and measures its log through the runtime (novox/hq ADR 0297): each verb is
// asked as mesh/log.<verb> naming the log, its answer is handed back whole, a key with no entry is nil,
// and a read pages on from Next until it is 0.
func TestABundleKeepsALogThroughTheRuntime(t *testing.T) {
	r := &logRuntime{}
	var (
		first, second, third uint64
		pages                []*LogPage
		last, none           *LogEntry
		info                 *LogInfo
		narrowed             *LogPage
	)
	got := logging(t, r, func() (any, error) {
		changes := Log("changes")
		var err error
		if first, err = changes.Append("401", map[string]any{"op": "opened", "by": "fix@g14"}); err != nil {
			return nil, err
		}
		if second, err = changes.Append("402", map[string]any{"op": "opened"}); err != nil {
			return nil, err
		}
		if third, err = changes.AppendAt("401", map[string]any{"op": "moved", "to": "fixing"}, first); err != nil {
			return nil, err
		}
		for _, k := range []string{"403", "404", "405"} {
			if _, err = changes.Append(k, map[string]any{"op": "opened"}); err != nil {
				return nil, err
			}
		}
		for from := uint64(1); ; {
			p, err := changes.Read(from, "", 2)
			if err != nil {
				return nil, err
			}
			pages = append(pages, p)
			if p.Next == 0 {
				break
			}
			from = p.Next
		}
		if narrowed, err = changes.Read(0, "401", 0); err != nil {
			return nil, err
		}
		if last, err = changes.Last("401"); err != nil {
			return nil, err
		}
		if none, err = changes.Last("999"); err != nil {
			return nil, err
		}
		info, err = changes.Info()
		return "done", err
	})
	if out, isError := text(t, got); isError || out != `"done"` {
		t.Fatalf("the tool answered %s", out)
	}
	if first != 1 || second != 2 || third != 3 {
		t.Fatalf("appended at %d, %d, %d", first, second, third)
	}

	// Asked as the contract names them: the verb, the log, and only what was given.
	method, p := r.params(t, 0)
	if method != "mesh/log.append" || p["log"] != "changes" || p["key"] != "401" || p["value"].(map[string]any)["by"] != "fix@g14" {
		t.Fatalf("an append asked %s %v", method, p)
	}
	if _, has := p["expect"]; has {
		t.Fatalf("an Append without expect named one: %v", p)
	}
	if method, p = r.params(t, 2); method != "mesh/log.append" || p["expect"] != float64(1) {
		t.Fatalf("an AppendAt asked %s %v", method, p)
	}
	if method, p = r.params(t, 6); method != "mesh/log.read" || p["from"] != float64(1) || p["limit"] != float64(2) || p["keys"] != nil {
		t.Fatalf("a read asked %s %v", method, p)
	}

	// Paged two at a time through six entries: three pages, every entry once, in order.
	if len(pages) != 3 {
		t.Fatalf("read %d pages, not 3", len(pages))
	}
	var seqs []uint64
	for _, pg := range pages {
		if pg.Last != 6 {
			t.Fatalf("a page said the last sequence is %d", pg.Last)
		}
		for _, e := range pg.Entries {
			seqs = append(seqs, e.Sequence)
		}
	}
	if fmt.Sprint(seqs) != "[1 2 3 4 5 6]" || pages[0].Next != 3 || pages[2].Next != 0 {
		t.Fatalf("paged %v, next %d then %d", seqs, pages[0].Next, pages[2].Next)
	}
	if e := pages[1].Entries[0]; e.Key != "401" || string(e.Value) != `{"op":"moved","to":"fixing"}` ||
		!e.At.Equal(time.Date(2026, 10, 11, 18, 0, 3, 0, time.UTC)) {
		t.Fatalf("an entry read as %+v", e)
	}

	// Narrowed to a key, with the runtime's page: no limit sent.
	method, p = r.params(t, 9)
	if method != "mesh/log.read" || p["keys"] != "401" || p["limit"] != nil {
		t.Fatalf("a narrowed read asked %s %v", method, p)
	}
	if len(narrowed.Entries) != 2 || narrowed.Entries[0].Sequence != 1 || narrowed.Entries[1].Sequence != 3 || narrowed.Next != 0 {
		t.Fatalf("narrowed to 401: %+v", narrowed)
	}

	if last == nil || last.Sequence != 3 || last.Key != "401" {
		t.Fatalf("the last of 401 is %+v", last)
	}
	if method, p = r.params(t, 10); method != "mesh/log.last" || p["key"] != "401" {
		t.Fatalf("a last asked %s %v", method, p)
	}
	if none != nil {
		t.Fatalf("a key with no entry has a last: %+v", none)
	}
	if info == nil || info.Entries != 6 || info.MaxBytes != 1<<30 || info.First != 1 || info.Last != 6 || info.Bytes == 0 {
		t.Fatalf("the log's info is %+v", info)
	}
	if method, p = r.params(t, 12); method != "mesh/log.info" || p["log"] != "changes" || len(p) != 1 {
		t.Fatalf("an info asked %s %v", method, p)
	}
}

// An AppendAt on a stale expect is ErrChanged and writes nothing: another entry for the key came in
// between. Expect 0 is a key with no entry yet, and is ErrChanged once the key has one. A log not issued
// is refused with the runtime's reason, and is not ErrChanged.
func TestAnAppendAtAStaleSequenceIsErrChangedAndWritesNothing(t *testing.T) {
	r := &logRuntime{}
	var errs []error
	got := logging(t, r, func() (any, error) {
		changes := Log("changes")
		read, err := changes.AppendAt("401", map[string]any{"op": "opened"}, 0)
		if err != nil {
			return nil, err
		}
		// Another writer is first.
		if _, err := changes.Append("401", map[string]any{"op": "noted"}); err != nil {
			return nil, err
		}
		_, stale := changes.AppendAt("401", map[string]any{"op": "moved"}, read)
		_, again := changes.AppendAt("401", map[string]any{"op": "opened"}, 0)
		_, other := Log("elsewhere").Append("401", map[string]any{})
		errs = []error{stale, again, other}
		return "done", nil
	})
	if out, isError := text(t, got); isError || out != `"done"` {
		t.Fatalf("the tool answered %s", out)
	}
	if !errors.Is(errs[0], ErrChanged) || !errors.Is(errs[1], ErrChanged) {
		t.Fatalf("a stale expect answered %v and %v, not ErrChanged", errs[0], errs[1])
	}
	if errs[2] == nil || errors.Is(errs[2], ErrChanged) || !strings.Contains(errs[2].Error(), "no log elsewhere") {
		t.Fatalf("a log not issued answered %v", errs[2])
	}
	if len(r.entries) != 2 {
		t.Fatalf("the log holds %d entries, not 2: a refused append wrote", len(r.entries))
	}
}

// A read asking for fewer than no entries is refused before the runtime is asked.
func TestAReadOfANegativeLimitIsRefused(t *testing.T) {
	r := &logRuntime{}
	var err error
	logging(t, r, func() (any, error) {
		_, err = Log("changes").Read(0, "", -1)
		return "done", nil
	})
	if err == nil || len(r.asked) != 0 {
		t.Fatalf("a negative limit answered %v after %d asks", err, len(r.asked))
	}
}