mesh-controller/internal/link/calls_test.go

package link

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"log"
	"strings"
	"sync"
	"testing"
	"time"
)

// answers collects what a call answered, and fails a second answer: the bus permits one.
type answers struct {
	t    *testing.T
	mu   sync.Mutex
	got  [][]byte
	sent chan struct{}
}

func newAnswers(t *testing.T) *answers { return &answers{t: t, sent: make(chan struct{}, 4)} }

func (a *answers) respond(body []byte) error {
	a.mu.Lock()
	defer a.mu.Unlock()
	a.got = append(a.got, body)
	if len(a.got) > 1 {
		a.t.Errorf("a call was answered %d times; the bus refuses every answer after the first", len(a.got))
	}
	a.sent <- struct{}{}
	return nil
}

func (a *answers) only() map[string]any {
	a.mu.Lock()
	defer a.mu.Unlock()
	if len(a.got) != 1 {
		a.t.Fatalf("%d answers, want exactly one", len(a.got))
	}
	var out map[string]any
	if err := json.Unmarshal(a.got[0], &out); err != nil {
		a.t.Fatal(err)
	}
	return out
}

func shortWindow(t *testing.T, d time.Duration) {
	t.Helper()
	was := AnswerWithin
	AnswerWithin = d
	t.Cleanup(func() { AnswerWithin = was })
}

// A call that finishes in time answers in full, once, and is kept as answered.
func TestACallThatFinishesInTimeAnswersInFull(t *testing.T) {
	l, a := NewCallLog(), newAnswers(t)
	l.serveCall("mesh-controller", "status", nil, "_INBOX.x.1", func(context.Context, json.RawMessage) (any, error) {
		return "all well", nil
	}, a.respond, nil)
	if got := a.only()["result"]; got != "all well" {
		t.Fatalf("answered %v", got)
	}
	recent := l.Recent()
	if len(recent) != 1 || recent[0].State != CallAnswered || string(recent[0].Args) != "{}" {
		t.Fatalf("kept %+v", recent)
	}
}

// **A call that outlasts AnswerWithin is answered that it is running, with its id** — before its
// caller gives up — and what it finally answered is kept under that id, not dropped (issue 265).
func TestACallThatOutlastsTheWindowSaysItIsRunningAndKeepsItsAnswer(t *testing.T) {
	shortWindow(t, 20*time.Millisecond)
	l, a := NewCallLog(), newAnswers(t)
	var logged bytes.Buffer
	release := make(chan struct{})
	finished := make(chan struct{})
	go func() {
		l.serveCall("mesh-controller", "push", json.RawMessage(`{"node":"anchor"}`), "_INBOX.x.2",
			func(context.Context, json.RawMessage) (any, error) {
				<-release
				return "anchor told", nil
			}, a.respond, log.New(&logged, "", 0))
		close(finished)
	}()
	<-a.sent
	got := a.only()["result"].(map[string]any)
	id, _ := got["call"].(string)
	if got["running"] != true || id == "" || !strings.Contains(got["output"].(string), id) {
		t.Fatalf("the running answer does not name its call: %v", got)
	}
	if c, _ := l.Get(id); c.State != CallRunning {
		t.Fatalf("while it runs it is kept as %q", c.State)
	}
	close(release)
	<-finished
	c, ok := l.Get(id)
	if !ok || c.State != CallFinishedAfter || !strings.Contains(string(c.Answer), "anchor told") {
		t.Fatalf("its answer was not kept: %+v", c)
	}
	if !strings.Contains(logged.String(), id) {
		t.Errorf("finishing late was not said: %q", logged.String())
	}
}

// A handler that acknowledges is answered then, not when it ends: a push answers before it sends.
func TestAnAcknowledgedCallIsAnsweredBeforeItGoesOn(t *testing.T) {
	l, a := NewCallLog(), newAnswers(t)
	done := make(chan struct{})
	go func() {
		l.serveCall("mesh-controller", "push", nil, "_INBOX.x.3", func(ctx context.Context, _ json.RawMessage) (any, error) {
			Acknowledge(ctx)
			Acknowledge(ctx) // a second time is nothing
			select {
			case <-a.sent: // the caller was answered before the work goes on
			case <-time.After(5 * time.Second):
				t.Error("acknowledging did not answer the caller")
			}
			return "sent", nil
		}, a.respond, nil)
		close(done)
	}()
	<-done
	if got := a.only()["result"].(map[string]any); got["running"] != true {
		t.Fatalf("an acknowledged call answered %v", got)
	}
	if c := l.Recent()[0]; c.State != CallFinishedAfter || !strings.Contains(string(c.Answer), "sent") {
		t.Fatalf("kept %+v", c)
	}
}

// **A refused answer is recorded against its call, in the mesh's words** — not only as the client
// library's line on standard error, which is all there was on 2026-10-05.
func TestARefusedAnswerIsKeptAgainstItsCall(t *testing.T) {
	l, a := NewCallLog(), newAnswers(t)
	reply := "_INBOX.laptop.node-tools.jJ4DRJFnZYvkPUBKYwUG5v.LNRyztuX"
	l.serveCall("mesh-controller", "push", nil, reply, func(context.Context, json.RawMessage) (any, error) {
		return "told", nil
	}, a.respond, nil)
	var logged bytes.Buffer
	refusal := errors.New(`nats: permissions violation: Permissions Violation for Publish to "` + reply + `" on connection [838]`)
	if !l.Refusal(refusal, log.New(&logged, "", 0)) {
		t.Fatal("the refusal of a kept call's answer was not recognised")
	}
	c := l.Recent()[0]
	if c.Refused == "" || !strings.Contains(logged.String(), c.ID) {
		t.Fatalf("the refusal is not kept or not said: %+v / %q", c, logged.String())
	}
	other := errors.New(`nats: permissions violation: Permissions Violation for Publish to "mesh.node.x" on connection [1]`)
	if l.Refusal(other, nil) || l.Refusal(errors.New("nats: timeout"), nil) {
		t.Error("an unrelated error was taken for a refused answer")
	}
}

// What `calls` keeps of the arguments never carries settings or a secret.
func TestACallKeepsNoSettingsOrSecrets(t *testing.T) {
	got := string(kept(json.RawMessage(`{"module":"m","values":"{\"token\":\"s3cret\"}","secret":"s3cret"}`)))
	if strings.Contains(got, "s3cret") || !strings.Contains(got, `"module":"m"`) {
		t.Fatalf("kept %s", got)
	}
}

// Only the newest KeptCalls are kept.
func TestTheLogKeepsTheNewest(t *testing.T) {
	l := NewCallLog()
	for i := 0; i < KeptCalls+5; i++ {
		l.begin("s", "v", nil, "")
	}
	recent := l.Recent()
	if len(recent) != KeptCalls || !strings.HasSuffix(recent[0].ID, "-105") || !strings.HasSuffix(recent[KeptCalls-1].ID, "-6") {
		t.Fatalf("kept %d, newest %s, oldest %s", len(recent), recent[0].ID, recent[len(recent)-1].ID)
	}
}